fix: optimize csv skill

packages/dbgpt-core/src/dbgpt/agent/skill/manage.py

@@ -419,7 +419,9 @@ class SkillManager(BaseComponent):
             ]
 
             if error_output:
-                chunks.append({"output_type": "text", "content": f"Error: {error_output}"})
+                chunks.append(
+                    {"output_type": "text", "content": f"Error: {error_output}"}
+                )
 
             if output:
                 chunks.append({"output_type": "text", "content": output})
@@ -438,7 +440,9 @@ class SkillManager(BaseComponent):
                 ensure_ascii=False,
             )
 
-    def get_skill_script_file(self, skill_name: str, script_file_name: str) -> Optional[str]:
+    def get_skill_script_file(
+        self, skill_name: str, script_file_name: str
+    ) -> Optional[str]:
         """Read a script file from skill's scripts directory.
 
         Args:
@@ -481,7 +485,9 @@ class SkillManager(BaseComponent):
 
         return references
 
-    def get_skill_reference_file(self, skill_name: str, ref_file_name: str) -> Optional[str]:
+    def get_skill_reference_file(
+        self, skill_name: str, ref_file_name: str
+    ) -> Optional[str]:
         """Read a specific reference file from skill's references directory.
 
         Args:
@@ -523,6 +529,18 @@ class SkillManager(BaseComponent):
         if not skills_dir:
             skills_dir = "skills"
 
+        # Search candidate subdirectories: direct, user/, claude/, project/, etc.
+        subdirs = ["", "user", "claude", "project"]
+        for subdir in subdirs:
+            candidate = (
+                os.path.join(skills_dir, subdir, skill_name)
+                if subdir
+                else os.path.join(skills_dir, skill_name)
+            )
+            if os.path.isdir(candidate):
+                return candidate
+
+        # Fallback: return direct path even if it doesn't exist yet
         return os.path.join(skills_dir, skill_name)
 
     async def get_skill_resource(
@@ -553,7 +571,16 @@ class SkillManager(BaseComponent):
         import os
 
         # Image file extensions that are not supported
-        IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp", ".svg", ".ico"}
+        IMAGE_EXTENSIONS = {
+            ".png",
+            ".jpg",
+            ".jpeg",
+            ".gif",
+            ".bmp",
+            ".webp",
+            ".svg",
+            ".ico",
+        }
 
         # Normalize the path
         resource_path = resource_path.lstrip("/\\")
@@ -585,7 +612,10 @@ class SkillManager(BaseComponent):
             real_skill_path = os.path.realpath(skill_path)
             if not real_path.startswith(real_skill_path):
                 return json.dumps(
-                    {"error": True, "message": f"Invalid resource path: {resource_path}"},
+                    {
+                        "error": True,
+                        "message": f"Invalid resource path: {resource_path}",
+                    },
                     ensure_ascii=False,
                 )
         except Exception as e:
@@ -601,13 +631,19 @@ class SkillManager(BaseComponent):
         # Otherwise, read the file content
         if not os.path.exists(full_path):
             return json.dumps(
-                {"error": True, "message": f"Resource '{resource_path}' not found in skill '{skill_name}'"},
+                {
+                    "error": True,
+                    "message": f"Resource '{resource_path}' not found in skill '{skill_name}'",
+                },
                 ensure_ascii=False,
             )
 
         if os.path.isdir(full_path):
             return json.dumps(
-                {"error": True, "message": f"'{resource_path}' is a directory, not a file"},
+                {
+                    "error": True,
+                    "message": f"'{resource_path}' is a directory, not a file",
+                },
                 ensure_ascii=False,
             )
 
@@ -624,7 +660,10 @@ class SkillManager(BaseComponent):
             )
         except UnicodeDecodeError:
             return json.dumps(
-                {"error": True, "message": f"Cannot read '{resource_path}': binary file not supported"},
+                {
+                    "error": True,
+                    "message": f"Cannot read '{resource_path}': binary file not supported",
+                },
                 ensure_ascii=False,
             )
         except Exception as e:
@@ -697,9 +736,7 @@ class SkillManager(BaseComponent):
 
         func_node = func_defs[main_func_name]
         # Get parameter names (skip 'self' for methods)
-        param_names = [
-            arg.arg for arg in func_node.args.args if arg.arg != "self"
-        ]
+        param_names = [arg.arg for arg in func_node.args.args if arg.arg != "self"]
 
         if len(param_names) == 1:
             param_name = param_names[0]
@@ -735,7 +772,14 @@ class SkillManager(BaseComponent):
 
         if not os.path.exists(script_path):
             return json.dumps(
-                {"chunks": [{"output_type": "text", "content": f"Script file not found: {script_path}"}]},
+                {
+                    "chunks": [
+                        {
+                            "output_type": "text",
+                            "content": f"Script file not found: {script_path}",
+                        }
+                    ]
+                },
                 ensure_ascii=False,
             )
 
@@ -744,7 +788,14 @@ class SkillManager(BaseComponent):
                 code = f.read()
         except Exception as e:
             return json.dumps(
-                {"chunks": [{"output_type": "text", "content": f"Error reading script: {str(e)}"}]},
+                {
+                    "chunks": [
+                        {
+                            "output_type": "text",
+                            "content": f"Error reading script: {str(e)}",
+                        }
+                    ]
+                },
                 ensure_ascii=False,
             )
 
@@ -754,17 +805,18 @@ class SkillManager(BaseComponent):
         if language == "python":
             adapted_args = self._adapt_args_for_script(code, args)
             args_repr = repr(adapted_args)
-            wrapper_code = f'''import sys
+            wrapper_code = f"""import sys
 import json
 
 sys.argv = ["script", json.dumps({args_repr})]
 __name__ = "__main__"
 
 {code}
-'''
+"""
             exec_code = wrapper_code
         else:
             from string import Template
+
             template = Template(code)
             exec_code = template.safe_substitute(**args)
 
@@ -772,21 +824,39 @@ __name__ = "__main__"
             code_server = await get_code_server(self.system_app)
             result = await code_server.exec(exec_code, language)
 
-            logs = result.logs.decode("utf-8") if isinstance(result.logs, bytes) else str(result.logs or "")
+            logs = (
+                result.logs.decode("utf-8")
+                if isinstance(result.logs, bytes)
+                else str(result.logs or "")
+            )
             exit_code = result.exit_code
 
             chunks = []
             if logs:
                 chunks.append({"output_type": "text", "content": logs})
             if exit_code != 0:
-                chunks.append({"output_type": "text", "content": f"Exit code: {exit_code}"})
+                chunks.append(
+                    {"output_type": "text", "content": f"Exit code: {exit_code}"}
+                )
             if not chunks:
-                chunks.append({"output_type": "text", "content": "Script executed successfully (no output)"})
+                chunks.append(
+                    {
+                        "output_type": "text",
+                        "content": "Script executed successfully (no output)",
+                    }
+                )
 
             return json.dumps({"chunks": chunks}, ensure_ascii=False)
         except Exception as e:
             return json.dumps(
-                {"chunks": [{"output_type": "text", "content": f"Script execution failed: {str(e)}"}]},
+                {
+                    "chunks": [
+                        {
+                            "output_type": "text",
+                            "content": f"Script execution failed: {str(e)}",
+                        }
+                    ]
+                },
                 ensure_ascii=False,
             )
 
@@ -805,18 +875,34 @@ __name__ = "__main__"
         skill_path = self._get_skill_path(skill_name)
         if not skill_path:
             return json.dumps(
-                {"chunks": [{"output_type": "text", "content": f"Skill '{skill_name}' not found"}]},
+                {
+                    "chunks": [
+                        {
+                            "output_type": "text",
+                            "content": f"Skill '{skill_name}' not found",
+                        }
+                    ]
+                },
                 ensure_ascii=False,
             )
 
         script_file_name = script_file_name.lstrip("/\\")
-        if script_file_name.startswith("scripts/") or script_file_name.startswith("scripts\\"):
+        if script_file_name.startswith("scripts/") or script_file_name.startswith(
+            "scripts\\"
+        ):
             script_file_name = script_file_name[8:]
 
         script_path = os.path.join(skill_path, "scripts", script_file_name)
         if not os.path.exists(script_path):
             return json.dumps(
-                {"chunks": [{"output_type": "text", "content": f"Script file '{script_file_name}' not found"}]},
+                {
+                    "chunks": [
+                        {
+                            "output_type": "text",
+                            "content": f"Script file '{script_file_name}' not found",
+                        }
+                    ]
+                },
                 ensure_ascii=False,
             )
 
@@ -825,18 +911,19 @@ __name__ = "__main__"
 
         adapted_args = self._adapt_args_for_script(code, args)
         args_repr = repr(adapted_args)
-        wrapper_code = f'''import sys
+        wrapper_code = f"""import sys
 import json
 
 sys.argv = ["script", json.dumps({args_repr})]
 __name__ = "__main__"
 
 {code}
-'''
+"""
 
         try:
             import sys
             import tempfile
+
             _IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"}
 
             scripts_dir = os.path.dirname(script_path)
@@ -868,9 +955,7 @@ __name__ = "__main__"
                     cwd=work_dir,
                     env=env,
                 )
-                stdout, stderr = await asyncio.wait_for(
-                    proc.communicate(), timeout=120
-                )
+                stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)
                 output_text = stdout.decode("utf-8", errors="replace")
                 error_text = stderr.decode("utf-8", errors="replace")
                 exit_code = proc.returncode or 0
@@ -882,13 +967,37 @@ __name__ = "__main__"
                     pass
             chunks = []
             if output_text.strip():
-                chunks.append({"output_type": "text", "content": output_text.strip()})
+                # If the script's stdout is already a valid JSON {"chunks": [...]}
+                # structure, use it directly to avoid double-encoding the content
+                # (which would cause JSON string values like CHART_DATA_JSON to
+                # have their quotes escaped as \" when later injected into HTML).
+                try:
+                    parsed_output = json.loads(output_text.strip())
+                    if isinstance(parsed_output, dict) and "chunks" in parsed_output:
+                        chunks = parsed_output["chunks"]
+                    else:
+                        chunks.append(
+                            {"output_type": "text", "content": output_text.strip()}
+                        )
+                except (json.JSONDecodeError, ValueError):
+                    chunks.append(
+                        {"output_type": "text", "content": output_text.strip()}
+                    )
             if exit_code != 0 and error_text.strip():
-                chunks.append({"output_type": "text", "content": f"[ERROR] {error_text.strip()}"})
+                chunks.append(
+                    {"output_type": "text", "content": f"[ERROR] {error_text.strip()}"}
+                )
             if exit_code != 0:
-                chunks.append({"output_type": "text", "content": f"Exit code: {exit_code}"})
+                chunks.append(
+                    {"output_type": "text", "content": f"Exit code: {exit_code}"}
+                )
             if not chunks:
-                chunks.append({"output_type": "text", "content": "Script executed successfully (no output)"})
+                chunks.append(
+                    {
+                        "output_type": "text",
+                        "content": "Script executed successfully (no output)",
+                    }
+                )
             # Scan work_dir for NEW image files generated by this run.
             # Return their absolute paths so the caller can copy them
             # to the static serving directory.
@@ -897,19 +1006,35 @@ __name__ = "__main__"
                     ext = os.path.splitext(fname)[1].lower()
                     full_path = os.path.join(root, fname)
                     if ext in _IMAGE_EXTS and full_path not in pre_existing_images:
-                        chunks.append({
-                            "output_type": "image",
-                            "content": full_path,
-                        })
+                        chunks.append(
+                            {
+                                "output_type": "image",
+                                "content": full_path,
+                            }
+                        )
             return json.dumps({"chunks": chunks}, ensure_ascii=False)
         except asyncio.TimeoutError:
             return json.dumps(
-                {"chunks": [{"output_type": "text", "content": "Script execution timed out (120s limit)"}]},
+                {
+                    "chunks": [
+                        {
+                            "output_type": "text",
+                            "content": "Script execution timed out (120s limit)",
+                        }
+                    ]
+                },
                 ensure_ascii=False,
             )
         except Exception as e:
             return json.dumps(
-                {"chunks": [{"output_type": "text", "content": f"Script execution failed: {str(e)}"}]},
+                {
+                    "chunks": [
+                        {
+                            "output_type": "text",
+                            "content": f"Script execution failed: {str(e)}",
+                        }
+                    ]
+                },
                 ensure_ascii=False,
             )
 

skills/csv-data-analysis/SKILL.md

@@ -5,7 +5,9 @@ description: This skill should be used when users need to analyze CSV files, und
 
 # 智能 CSV 数据深度分析工具
 
-CSV数据分析工具是一个基于 AI 与前端可视化技术（ECharts + Tailwind CSS）的深度自动化数据探索工具。它能够快速提取统计特征、分类信息、相关性以及时序趋势，并由大模型注入深度业务洞察，生成高度美观和可交互的网页分析报告。
+CSV数据分析工具是一个基于 AI 与前端可视化技术（ECharts + Tailwind CSS）的深度自动化数据探索工具。它能够快速提取统计特征、数据质量、数值分布、异常值检测、分类信息、相关性、排名以及时序趋势，并在后半段补充异动概述、归因线索和总结建议，生成高度美观和可交互的网页分析报告。
+
+报告整体遵循“前半段基础数据分析、后半段异动与归因增强”的结构，核心章节包括：报告摘要、数据概览与质量检查、数值指标分布特征、特征分析与结构分析、关系分析与异常识别、数据异动概述、归因分析模块、分析结果与统计明细、原因推测/总结/建议。
 
 ## 核心工作流（LLM 必读）
 
@@ -27,7 +29,7 @@ CSV数据分析工具是一个基于 AI 与前端可视化技术（ECharts + Tai
 **脚本返回说明：**
 脚本会返回一大段 `text` 内容，其中包含两个部分：
 1. **【统计摘要】**：供你阅读并理解数据集的基本情况、分布、相关性和分类构成。
-2. **【CHART_DATA_JSON】**：位于 `###CHART_DATA_JSON_START###` 和 `###CHART_DATA_JSON_END###` 之间的纯 JSON 字符串。这是用于渲染交互式图表的原生数据。
+2. **【marker 包裹的数据块】**：脚本输出里会带有 `###KEY_START###...###KEY_END###` 形式的 marker 数据块。后端会自动捕获并注入到模板中，**你不需要关心这部分内容，也不需要传递它**。
 
 ### 第二步：生成洞察与展示报告 (注入模板)
 
@@ -35,15 +37,19 @@ CSV数据分析工具是一个基于 AI 与前端可视化技术（ECharts + Tai
 
 **关键规则（必须遵守）：**
 
-1. **必须设置 `template_path`** 为 `csv-data-analysis/templates/report_template.html`。模板中已内置完整的 ECharts 渲染 JavaScript 代码和所有章节标题、页脚文本，你只需要通过 `data` 参数填充 9 个内容占位符即可。**绝对不要自己编写或修改任何 JavaScript 图表渲染代码。**
+1. **必须设置 `template_path`** 为 `csv-data-analysis/templates/report_template.html`。模板中已内置完整的 ECharts 渲染 JavaScript 代码和所有章节标题、页脚文本，你只需要通过 `data` 参数填充 8 个内容占位符即可。**绝对不要自己编写或修改任何 JavaScript 图表渲染代码。**
 
-2. **`CHART_DATA_JSON`** 必须**完整且原封不动**地复制自脚本输出中 `###CHART_DATA_JSON_START###` 和 `###CHART_DATA_JSON_END###` 之间的纯 JSON 字符串。不要自行编造，不需要做任何转义处理。
+2. **marker 数据块由后端自动注入**，你无需也不应在 `data` 中传递它。后端会从脚本输出里的 `###KEY_START###...###KEY_END###` 自动提取并注入到模板；当前这个 skill 中主要是 `CHART_DATA_JSON`。
 
 3. **`*_INSIGHTS`、`EXEC_SUMMARY` 和 `CONCLUSIONS`** 必须使用 HTML 格式（如 `<p>`, `<ul>`, `<li>`, `<strong>`, `<ol>`）来确保排版美观。这些内容由你基于统计摘要撰写深度业务洞察。
 
 4. **输出语言必须与用户输入语言一致。**
 
-5. **只传 9 个占位符，不要多也不要少。** 模板已将所有章节标题（Distribution Analysis、Correlation Analysis 等）、洞察框标题（Insights）和页脚文本硬编码在 HTML 中，你无需传递这些。
+5. **只传 8 个占位符，不要多也不要少。** `CHART_DATA_JSON` 这类 marker 自动注入字段由后端处理，不需要你传递。模板已将所有章节标题（Distribution Analysis、Correlation Analysis 等）、洞察框标题（Insights）和页脚文本硬编码在 HTML 中，你无需传递这些。
+
+6. **洞察内容必须更充实。** 每个洞察模块尽量覆盖 4 层信息：`现象`、`可能原因`、`业务影响`、`行动建议`。不要只复述统计值，也不要只写一两句空泛结论。
+
+7. **基础分析优先，归因为增强模块。** 报告前半段必须重点分析 CSV 本身的数据特征，包括数值分布、分类结构、异常值、相关关系、排序特征等，并尽量结合图表解读；“数据异动概述”“归因分析”“原因推测”应放在后半段作为增强模块，不能让整份报告只剩归因内容。
 
 **`html_interpreter` 调用示例：**
 ```json
@@ -57,36 +63,34 @@ CSV数据分析工具是一个基于 AI 与前端可视化技术（ECharts + Tai
     "CORRELATION_INSIGHTS": "<p>变量间的热力图揭示了强烈的正相关关系，特别是...，这意味着...</p>",
     "CATEGORICAL_INSIGHTS": "<p>分类占比显示，'城市'字段中北京与上海占据了 50% 以上的份额。</p>",
     "TIME_SERIES_INSIGHTS": "<p>从时序趋势中可以看出，数据在年末存在显著的季节性拉升现象。</p>",
-    "CONCLUSIONS": "<p>综合以上多维度分析，数据呈现出明确的结构性特征与规律。</p><h3>建议</h3><ul><li>建议定期检查缺失值比例...</li><li>重点关注高增长细分市场...</li></ul>",
-    "CHART_DATA_JSON": "此处粘贴脚本输出中 ###CHART_DATA_JSON_START### 到 ###CHART_DATA_JSON_END### 之间的完整 JSON 字符串"
+    "CONCLUSIONS": "<p>综合以上多维度分析，数据呈现出明确的结构性特征与规律。</p><h3>建议</h3><ul><li>建议定期检查缺失值比例...</li><li>重点关注高增长细分市场...</li></ul>"
   }
 }
 ```
 
 > **严禁事项：**
+> - 禁止在 `data` 中传递 `CHART_DATA_JSON` 或任何 marker 自动注入字段（后端自动处理）
 > - 禁止在 `data` 中添加任何 JavaScript 代码
 > - 禁止省略 `template_path` 参数（不设置 template_path 会导致图表无法渲染！）
-> - 禁止对 `CHART_DATA_JSON` 做任何修改、截断或二次编码
 > - 禁止返回静态 PNG 图片，本工具已全面升级为 ECharts 动态前端渲染
-> - 禁止传递不存在的占位符（模板只有以下 9 个占位符，传递其他名称会被忽略）
+> - 禁止传递不存在的占位符（模板只有以下 8 个文本占位符 + 1 个自动注入的 CHART_DATA_JSON，传递其他名称会被忽略）
 
-## 占位符清单（共 9 个）
+## 占位符清单（共 8 个，由 LLM 通过 data 传递）
 
-模板中的全部占位符如下（通过 `data` 参数传递）：
+模板中需要你填充的占位符如下：
 
 | 占位符 | 类型 | 必填 | 说明 |
 |---|---|---|---|
 | `REPORT_TITLE` | 文本 | 是 | 报告标题，如"销售数据集深度分析报告" |
 | `REPORT_SUBTITLE` | 文本 | 是 | 报告副标题，如"多维度数据特征与业务洞见挖掘" |
-| `EXEC_SUMMARY` | HTML | 是 | 执行摘要（你撰写的深度总结），使用 `<p>`, `<ul>`, `<li>` 等标签 |
-| `DISTRIBUTION_INSIGHTS` | HTML | 是 | 数值分布洞察正文 |
-| `CORRELATION_INSIGHTS` | HTML | 是 | 相关性分析洞察正文 |
-| `CATEGORICAL_INSIGHTS` | HTML | 是 | 分类变量洞察正文 |
-| `TIME_SERIES_INSIGHTS` | HTML | 是 | 时间序列洞察正文 |
-| `CONCLUSIONS` | HTML | 是 | 结论与建议正文（包含建议内容，直接写在此字段中） |
-| `CHART_DATA_JSON` | JSON字符串 | 是 | 脚本输出的原始 JSON（驱动图表渲染），必须从脚本输出原封不动复制 |
+| `EXEC_SUMMARY` | HTML | 是 | 报告摘要：概览数据规模、主要发现和结论预告 |
+| `DISTRIBUTION_INSIGHTS` | HTML | 是 | 数值指标分布特征解读：偏态、波动、分位区间、离散程度 |
+| `CORRELATION_INSIGHTS` | HTML | 是 | 关系分析与异常识别解读：相关性、联动、异常点、结构关系 |
+| `CATEGORICAL_INSIGHTS` | HTML | 是 | 特征分析与结构分析解读：分类结构、集中度、排名和分组特征 |
+| `TIME_SERIES_INSIGHTS` | HTML | 是 | 数据异动概述部分的补充解读：若有时间列则讲趋势；若无时间列则讲分层差异与异动概况 |
+| `CONCLUSIONS` | HTML | 是 | 原因推测、总结与建议正文；要区分“数据证据”和“合理推测” |
 
-> **注意：** 模板中的所有章节标题（如 "Distribution Analysis"、"Correlation Analysis"、"Conclusions & Recommendations" 等）、洞察框标题（"Insights"）和页脚文本已硬编码在 HTML 中，无需通过占位符传递。
+> **注意：** `csv_analyzer.py` 会在输出中附带 `###CHART_DATA_JSON_START###...###CHART_DATA_JSON_END###` marker 数据块，后端会自动提取并注入模板，无需在 `data` 中传递。模板中的所有章节标题（如 "Distribution Analysis"、"Correlation Analysis"、"Conclusions & Recommendations" 等）、洞察框标题（"Insights"）和页脚文本已硬编码在 HTML 中，无需通过占位符传递。
 
 ## 为什么选择本工具？
 

skills/csv-data-analysis/scripts/csv_analyzer.py

@@ -13,11 +13,96 @@ def log(*args, **kwargs):
     print(*args, file=sys.stderr, **kwargs)
 
 
+def safe_div(numerator, denominator):
+    if denominator in (0, None):
+        return 0.0
+    return float(numerator) / float(denominator)
+
+
+def classify_skewness(value):
+    abs_val = abs(value)
+    if abs_val >= 1:
+        return "明显偏态"
+    if abs_val >= 0.5:
+        return "中度偏态"
+    return "近似对称"
+
+
+def classify_cv(value):
+    if value >= 100:
+        return "极高波动"
+    if value >= 50:
+        return "高波动"
+    if value >= 20:
+        return "中等波动"
+    return "低波动"
+
+
+def select_primary_metric(df, numeric_cols):
+    if not numeric_cols:
+        return None
+
+    preferred_keywords = [
+        "score",
+        "value",
+        "amount",
+        "sales",
+        "revenue",
+        "profit",
+        "price",
+        "rate",
+        "index",
+        "metric",
+        "total",
+        "count",
+        "分数",
+        "得分",
+        "金额",
+        "销售",
+        "收入",
+        "利润",
+        "价格",
+        "指标",
+        "总量",
+        "数量",
+        "评分",
+    ]
+    skip_keywords = ["rank", "ranking", "id", "index_id", "序号", "排名", "编号"]
+
+    candidates = []
+    for col in numeric_cols:
+        col_lower = str(col).lower()
+        unique_cnt = int(df[col].nunique())
+        score = 0
+        if unique_cnt > 5:
+            score += 2
+        if not any(kw in col_lower for kw in skip_keywords):
+            score += 2
+        if any(kw in col_lower for kw in preferred_keywords):
+            score += 4
+        series = df[col].dropna()
+        if len(series) > 0:
+            score += 1
+            if float(series.std()) > 0:
+                score += 1
+        candidates.append((score, col))
+
+    candidates.sort(key=lambda x: x[0], reverse=True)
+    return candidates[0][1] if candidates else numeric_cols[0]
+
+
+def select_label_col(df):
+    for col in df.columns:
+        if df[col].dtype == "object" and df[col].nunique() > 1:
+            return col
+    return None
+
+
 def analyze_csv(file_path):
     """
     分析CSV文件，提取用于 ECharts 渲染的数据结构和用于 LLM 分析的统计摘要。
-    输出包含: overview, distributions, correlations, categories, time_series,
-    scatter (散点图), stats_table (统计表格)
+    输出包含: overview, data_quality, distributions, correlations, categories,
+    time_series, scatter, stats_table, box_plots, outliers, top_bottom
     """
     try:
         log(f"正在读取文件: {file_path}")
@@ -31,21 +116,64 @@ def analyze_csv(file_path):
         missing_pct = (
             round((missing_cells / total_cells) * 100, 2) if total_cells > 0 else 0
         )
+        duplicate_rows = int(df.duplicated().sum())
 
         overview = {
             "rows": int(df.shape[0]),
             "cols": int(df.shape[1]),
             "missing_cells": missing_cells,
             "missing_pct": missing_pct,
+            "duplicate_rows": duplicate_rows,
+            "memory_kb": round(df.memory_usage(deep=True).sum() / 1024, 1),
         }
 
+        # ==========================================
+        # 1b. 数据质量分析 (每列缺失率 + 数据类型)
+        # ==========================================
+        data_quality = {
+            "columns": [],
+            "missing_rates": [],
+            "dtypes": [],
+            "unique_counts": [],
+            "dtype_summary": {},
+        }
+        for col in df.columns:
+            data_quality["columns"].append(str(col))
+            col_missing = int(df[col].isnull().sum())
+            rate = round((col_missing / len(df)) * 100, 1) if len(df) > 0 else 0
+            data_quality["missing_rates"].append(rate)
+            data_quality["dtypes"].append(str(df[col].dtype))
+            data_quality["unique_counts"].append(int(df[col].nunique()))
+
+        # dtype breakdown for overview
+        dtype_counts = {}
+        for dt in data_quality["dtypes"]:
+            cat = (
+                "numeric"
+                if "int" in dt or "float" in dt
+                else ("datetime" if "datetime" in dt else "text")
+            )
+            dtype_counts[cat] = dtype_counts.get(cat, 0) + 1
+        data_quality["dtype_summary"] = dtype_counts
+        missing_by_col = sorted(
+            [
+                (col, rate, int(df[col].isnull().sum()))
+                for col, rate in zip(df.columns, data_quality["missing_rates"])
+            ],
+            key=lambda x: x[1],
+            reverse=True,
+        )
+
         # ==========================================
         # 2. 数值列分析 (直方图分布 & 相关性)
         # ==========================================
         numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
+        primary_metric = select_primary_metric(df, numeric_cols)
+        label_col = select_label_col(df)
         distributions = {}
         correlations = {"cols": numeric_cols, "data": []}
         numeric_summary = {}
+        correlation_highlights = {"positive": [], "negative": []}
 
         if numeric_cols:
             # 取最多前 8 个数值列画分布图
@@ -62,31 +190,58 @@ def analyze_csv(file_path):
                         "bins": bins,
                         "counts": [int(x) for x in hist],
                     }
+                    # Skewness & Kurtosis
+                    skew_val = float(s.skew()) if len(s) > 2 else 0.0
+                    kurt_val = float(s.kurtosis()) if len(s) > 3 else 0.0
+                    mean_val = float(s.mean())
+                    std_val = float(s.std())
+                    cv = (
+                        round(abs(std_val / mean_val) * 100, 1)
+                        if mean_val != 0
+                        else 0.0
+                    )
+                    spread = float(s.max()) - float(s.min())
                     numeric_summary[col] = {
                         "min": float(s.min()),
                         "max": float(s.max()),
-                        "mean": float(s.mean()),
+                        "mean": round(mean_val, 4),
                         "median": float(s.median()),
-                        "std": float(s.std()),
+                        "std": round(std_val, 4),
                         "q25": float(s.quantile(0.25)),
                         "q75": float(s.quantile(0.75)),
+                        "p5": float(s.quantile(0.05)),
+                        "p95": float(s.quantile(0.95)),
+                        "cv": cv,
+                        "spread": round(spread, 4),
+                        "skewness": round(skew_val, 3),
+                        "kurtosis": round(kurt_val, 3),
                     }
 
             # 相关性矩阵 (取全部数值列)
             if len(numeric_cols) > 1:
                 corr_df = df[numeric_cols].corr(method="pearson").fillna(0).round(2)  # type: ignore[call-overload]
+                corr_pairs = []
                 for i, col1 in enumerate(numeric_cols):
                     for j, col2 in enumerate(numeric_cols):
                         correlations["data"].append([i, j, float(corr_df.iloc[i, j])])
+                        if i < j:
+                            corr_pairs.append((col1, col2, float(corr_df.iloc[i, j])))
+                corr_pairs = sorted(corr_pairs, key=lambda x: x[2], reverse=True)
+                correlation_highlights["positive"] = corr_pairs[:3]
+                correlation_highlights["negative"] = sorted(
+                    corr_pairs, key=lambda x: x[2]
+                )[:3]
 
         # ==========================================
-        # 3. 分类列分析 (饼图/柱状图)
+        # 3. 分类列分析 (饼图/柱状图 + 熵/集中度)
         # ==========================================
         categorical_cols = df.select_dtypes(
             include=["object", "category"]
         ).columns.tolist()
         categories = {}
         cat_summary = {}
+        segment_breakdown = []
+        segment_comparison = {}
 
         if categorical_cols:
             # 取最多前 6 个分类列
@@ -100,14 +255,79 @@ def analyze_csv(file_path):
                         }
                         top1 = val_counts.index[0]
                         top1_count = val_counts.values[0]
-                        cat_summary[col] = (
-                            f"唯一值数量: {df[col].nunique()}，最常见: {top1} (出现 {top1_count} 次)"
+                        n_unique = df[col].nunique()
+                        total_non_null = int(df[col].notna().sum())
+                        # Shannon entropy (log2)
+                        probs = np.array(val_counts.values, dtype=float)
+                        if total_non_null > 0:
+                            probs = probs / float(total_non_null)
+                        entropy = -float(np.sum(probs * np.log2(probs + 1e-12)))
+                        # Concentration ratio: top-3 share
+                        top3_share = (
+                            round(val_counts.head(3).sum() / total_non_null * 100, 1)
+                            if total_non_null > 0
+                            else 0
                         )
+                        cat_summary[col] = {
+                            "n_unique": n_unique,
+                            "top1": str(top1),
+                            "top1_count": int(top1_count),
+                            "top1_share": round(
+                                safe_div(top1_count, total_non_null) * 100, 1
+                            )
+                            if total_non_null > 0
+                            else 0,
+                            "entropy": round(entropy, 3),
+                            "top3_share": top3_share,
+                        }
+
+            if primary_metric:
+                for col in categorical_cols[:3]:
+                    grouped = (
+                        df[[col, primary_metric]]
+                        .dropna()
+                        .groupby(col)[primary_metric]
+                        .agg(["count", "mean", "sum"])
+                    )
+                    grp = (
+                        pd.DataFrame(grouped)
+                        .reset_index()
+                        .sort_values("sum", ascending=False)
+                        .head(5)
+                    )
+                    if not grp.empty:
+                        segment_breakdown.append(
+                            {
+                                "dimension": col,
+                                "metric": primary_metric,
+                                "leaders": [
+                                    {
+                                        "name": str(row[col]),
+                                        "count": int(row["count"]),
+                                        "mean": round(float(row["mean"]), 2),
+                                        "sum": round(float(row["sum"]), 2),
+                                    }
+                                    for _, row in grp.iterrows()
+                                ],
+                            }
+                        )
+                if segment_breakdown:
+                    lead_segment = segment_breakdown[0]
+                    leaders = lead_segment["leaders"][:8]
+                    segment_comparison = {
+                        "dimension": lead_segment["dimension"],
+                        "metric": lead_segment["metric"],
+                        "labels": [item["name"] for item in leaders],
+                        "values": [item["mean"] for item in leaders],
+                        "counts": [item["count"] for item in leaders],
+                    }
 
         # ==========================================
-        # 4. 时间序列分析
+        # 4. 时间序列分析 (支持多个数值列)
         # ==========================================
         time_series = {"name": "", "dates": [], "values": []}
+        time_series_multi = []  # 额外的时序列数据
+        time_series_diagnostics = {}
 
         if numeric_cols:
             date_col = None
@@ -121,17 +341,19 @@ def analyze_csv(file_path):
                         pass
 
             if date_col:
-                num_col = numeric_cols[0]
                 df_ts = df.copy()
                 df_ts[date_col] = pd.to_datetime(df_ts[date_col], errors="coerce")
-                df_ts = df_ts.dropna(subset=[date_col, num_col])
+                df_ts = df_ts.dropna(subset=[date_col])
 
-                if not df_ts.empty:
-                    df_ts = df_ts.set_index(date_col)
+                # 主时序：第一个数值列
+                num_col = primary_metric or numeric_cols[0]
+                df_ts_main = df_ts.dropna(subset=[num_col]).copy()
+                if not df_ts_main.empty:
+                    df_ts_main = df_ts_main.set_index(date_col)
                     try:
-                        monthly = df_ts[num_col].resample("M").mean().dropna()
+                        monthly = df_ts_main[num_col].resample("M").mean().dropna()
                         if len(monthly) < 3:
-                            monthly = df_ts[num_col].resample("D").mean().dropna()
+                            monthly = df_ts_main[num_col].resample("D").mean().dropna()
                         monthly = monthly.tail(100)
 
                         time_series["name"] = num_col
@@ -141,15 +363,100 @@ def analyze_csv(file_path):
                         time_series["values"] = [
                             round(float(x), 2) for x in monthly.values
                         ]
+                        if len(monthly) >= 2:
+                            idx = np.arange(len(monthly), dtype=float)
+                            slope = float(
+                                np.polyfit(idx, monthly.values.astype(float), 1)[0]
+                            )
+                            first_val = float(monthly.iloc[0])
+                            last_val = float(monthly.iloc[-1])
+                            peak_idx = int(np.argmax(monthly.values))
+                            trough_idx = int(np.argmin(monthly.values))
+                            pct_change = (
+                                safe_div(last_val - first_val, abs(first_val)) * 100
+                            )
+                            time_series_diagnostics = {
+                                "date_col": date_col,
+                                "metric": num_col,
+                                "points": int(len(monthly)),
+                                "start": round(first_val, 2),
+                                "end": round(last_val, 2),
+                                "change_pct": round(pct_change, 1),
+                                "slope": round(slope, 4),
+                                "volatility_pct": round(
+                                    safe_div(monthly.std(), abs(monthly.mean())) * 100,
+                                    1,
+                                )
+                                if float(monthly.mean()) != 0
+                                else 0,
+                                "peak_date": monthly.index[peak_idx].strftime(
+                                    "%Y-%m-%d"
+                                ),
+                                "peak_value": round(float(monthly.iloc[peak_idx]), 2),
+                                "trough_date": monthly.index[trough_idx].strftime(
+                                    "%Y-%m-%d"
+                                ),
+                                "trough_value": round(
+                                    float(monthly.iloc[trough_idx]), 2
+                                ),
+                            }
                     except Exception as e:
                         log(f"时间序列处理失败: {e}")
 
+                # 额外时序列（最多再加2个）
+                for extra_col in numeric_cols[1:3]:
+                    df_ts_extra = df_ts.dropna(subset=[extra_col]).copy()
+                    if not df_ts_extra.empty:
+                        df_ts_extra = df_ts_extra.set_index(date_col)
+                        try:
+                            monthly_e = (
+                                df_ts_extra[extra_col].resample("M").mean().dropna()
+                            )
+                            if len(monthly_e) < 3:
+                                monthly_e = (
+                                    df_ts_extra[extra_col].resample("D").mean().dropna()
+                                )
+                            monthly_e = monthly_e.tail(100)
+                            if len(monthly_e) >= 2:
+                                time_series_multi.append(
+                                    {
+                                        "name": extra_col,
+                                        "dates": [
+                                            x.strftime("%Y-%m-%d")
+                                            for x in monthly_e.index
+                                        ],
+                                        "values": [
+                                            round(float(x), 2) for x in monthly_e.values
+                                        ],
+                                    }
+                                )
+                        except Exception:
+                            pass
+
         # ==========================================
         # 5. 散点图数据 (前两个数值列)
         # ==========================================
         scatter = {}
         if len(numeric_cols) >= 2:
-            col_x, col_y = numeric_cols[0], numeric_cols[1]
+            if primary_metric and correlations["data"]:
+                corr_candidates = []
+                for i, j, val in correlations["data"]:
+                    left = numeric_cols[i]
+                    right = numeric_cols[j]
+                    if left == right:
+                        continue
+                    if left == primary_metric:
+                        corr_candidates.append((abs(val), right, primary_metric))
+                    elif right == primary_metric:
+                        corr_candidates.append((abs(val), left, primary_metric))
+                corr_candidates.sort(key=lambda x: x[0], reverse=True)
+                if corr_candidates:
+                    _, partner, metric = corr_candidates[0]
+                    col_x, col_y = partner, metric
+                else:
+                    col_x, col_y = numeric_cols[0], numeric_cols[1]
+            else:
+                col_x, col_y = numeric_cols[0], numeric_cols[1]
             df_scatter = df[[col_x, col_y]].dropna()
             # 限制最多 500 个点，避免数据过大
             if len(df_scatter) > 500:
@@ -162,31 +469,227 @@ def analyze_csv(file_path):
             }
 
         # ==========================================
-        # 6. 统计汇总表格
+        # 5b. 箱线图数据 (Box Plot) — 前 8 个数值列
+        # ==========================================
+        box_plots = {}
+        if numeric_cols:
+            for col in numeric_cols[:8]:
+                s = df[col].dropna()
+                if len(s) > 0:
+                    q1 = float(s.quantile(0.25))
+                    q3 = float(s.quantile(0.75))
+                    iqr = q3 - q1
+                    lower_fence = q1 - 1.5 * iqr
+                    upper_fence = q3 + 1.5 * iqr
+                    outlier_vals = s[(s < lower_fence) | (s > upper_fence)]
+                    # Limit outlier points to 50 for rendering
+                    outlier_list = [
+                        round(float(v), 4)
+                        for v in outlier_vals.head(50).tolist()  # type: ignore[union-attr]
+                    ]
+                    box_plots[col] = {
+                        "min": round(float(s.min()), 4),
+                        "q1": round(q1, 4),
+                        "median": round(float(s.median()), 4),
+                        "q3": round(q3, 4),
+                        "max": round(float(s.max()), 4),
+                        "lower_fence": round(max(float(s.min()), lower_fence), 4),
+                        "upper_fence": round(min(float(s.max()), upper_fence), 4),
+                        "outliers": outlier_list,
+                    }
+
+        # ==========================================
+        # 5c. 异常值检测汇总 (IQR method)
+        # ==========================================
+        outliers = {}
+        if numeric_cols:
+            for col in numeric_cols[:8]:
+                s = df[col].dropna()
+                if len(s) > 0:
+                    q1 = float(s.quantile(0.25))
+                    q3 = float(s.quantile(0.75))
+                    iqr = q3 - q1
+                    lower = q1 - 1.5 * iqr
+                    upper = q3 + 1.5 * iqr
+                    n_outliers = int(((s < lower) | (s > upper)).sum())
+                    outliers[col] = {
+                        "count": n_outliers,
+                        "pct": round((n_outliers / len(s)) * 100, 1)
+                        if len(s) > 0
+                        else 0,
+                        "lower_bound": round(lower, 4),
+                        "upper_bound": round(upper, 4),
+                    }
+
+        # ==========================================
+        # 5d. Top/Bottom 排名 (数值列的 Top5 / Bottom5)
+        # ==========================================
+        top_bottom = {}
+        ranking_signal = {}
+        if numeric_cols and len(df) > 0:
+            rank_col = primary_metric or numeric_cols[0]
+
+            if rank_col:
+                df_sorted = df.dropna(subset=[rank_col]).sort_values(
+                    rank_col, ascending=False
+                )
+                top5 = df_sorted.head(5)
+                bottom5 = df_sorted.tail(5).iloc[::-1]  # reverse so worst first
+
+                def extract_ranked(subset):
+                    labels = []
+                    values = []
+                    for _, row in subset.iterrows():
+                        lbl = str(row[label_col])[:30] if label_col else str(row.name)
+                        labels.append(lbl)
+                        values.append(round(float(row[rank_col]), 2))
+                    return {"labels": labels, "values": values}
+
+                top_bottom = {
+                    "rank_col": rank_col,
+                    "label_col": label_col or "index",
+                    "top5": extract_ranked(top5),
+                    "bottom5": extract_ranked(bottom5),
+                }
+                if top_bottom["top5"]["values"] and top_bottom["bottom5"]["values"]:
+                    top_avg = float(np.mean(top_bottom["top5"]["values"]))
+                    bottom_avg = float(np.mean(top_bottom["bottom5"]["values"]))
+                    ranking_signal = {
+                        "top_avg": round(top_avg, 2),
+                        "bottom_avg": round(bottom_avg, 2),
+                        "gap": round(top_avg - bottom_avg, 2),
+                    }
+
+        # ==========================================
+        # 6b. 主指标异动概览与归因结构
+        # ==========================================
+        anomaly_overview = {}
+        driver_analysis = {"metric": "", "items": []}
+        if primary_metric and primary_metric in df.columns:
+            metric_series = df[primary_metric].dropna()
+            if len(metric_series) > 0:
+                q10 = float(metric_series.quantile(0.1))
+                q25 = float(metric_series.quantile(0.25))
+                q50 = float(metric_series.quantile(0.5))
+                q75 = float(metric_series.quantile(0.75))
+                q90 = float(metric_series.quantile(0.9))
+                band_labels = ["P0-P25", "P25-P50", "P50-P75", "P75-P100"]
+                band_values = [
+                    int((metric_series <= q25).sum()),
+                    int(((metric_series > q25) & (metric_series <= q50)).sum()),
+                    int(((metric_series > q50) & (metric_series <= q75)).sum()),
+                    int((metric_series > q75).sum()),
+                ]
+                top_group = df[df[primary_metric] >= q90]
+                bottom_group = df[df[primary_metric] <= q10]
+                primary_outlier = outliers.get(primary_metric, {})
+                anomaly_overview = {
+                    "metric": primary_metric,
+                    "mean": round(float(metric_series.mean()), 2),
+                    "median": round(float(metric_series.median()), 2),
+                    "std": round(float(metric_series.std()), 2),
+                    "q10": round(q10, 2),
+                    "q90": round(q90, 2),
+                    "top_group_size": int(len(top_group)),
+                    "bottom_group_size": int(len(bottom_group)),
+                    "top_group_mean": round(float(top_group[primary_metric].mean()), 2)
+                    if len(top_group) > 0
+                    else 0,
+                    "bottom_group_mean": round(
+                        float(bottom_group[primary_metric].mean()), 2
+                    )
+                    if len(bottom_group) > 0
+                    else 0,
+                    "gap": round(
+                        float(top_group[primary_metric].mean())
+                        - float(bottom_group[primary_metric].mean()),
+                        2,
+                    )
+                    if len(top_group) > 0 and len(bottom_group) > 0
+                    else 0,
+                    "band_labels": band_labels,
+                    "band_values": band_values,
+                    "outlier_count": int(primary_outlier.get("count", 0)),
+                    "outlier_pct": float(primary_outlier.get("pct", 0)),
+                }
+
+                driver_items = []
+                metric_std = float(metric_series.std()) if len(metric_series) > 1 else 0
+                for col in numeric_cols:
+                    if col == primary_metric:
+                        continue
+                    pair = df[[primary_metric, col]].dropna()
+                    if len(pair) < 5:
+                        continue
+                    metric_pair_series = pair.iloc[:, 0]
+                    col_pair_series = pair.iloc[:, 1]
+                    corr_val = float(metric_pair_series.corr(col_pair_series))
+                    top_mean = (
+                        float(top_group[col].mean())
+                        if len(top_group) > 0 and col in top_group.columns
+                        else 0
+                    )
+                    bottom_mean = (
+                        float(bottom_group[col].mean())
+                        if len(bottom_group) > 0 and col in bottom_group.columns
+                        else 0
+                    )
+                    col_std = float(col_pair_series.std()) if len(pair) > 1 else 0
+                    gap_ratio = safe_div(
+                        top_mean - bottom_mean, col_std if col_std else 1
+                    )
+                    score = round(
+                        min(100, abs(corr_val) * 55 + min(abs(gap_ratio), 3) * 15), 1
+                    )
+                    driver_items.append(
+                        {
+                            "name": col,
+                            "corr": round(corr_val, 3),
+                            "top_mean": round(top_mean, 2),
+                            "bottom_mean": round(bottom_mean, 2),
+                            "gap_ratio": round(gap_ratio, 2),
+                            "score": score,
+                        }
+                    )
+                driver_items.sort(key=lambda x: x["score"], reverse=True)
+                driver_analysis = {"metric": primary_metric, "items": driver_items[:8]}
+
+        # ==========================================
+        # 6. 统计汇总表格 (含新增 P5/P95/CV 列)
         # ==========================================
         stats_table = {"headers": [], "rows": []}
         if numeric_summary:
             stats_table["headers"] = [
                 "变量",
                 "最小值",
+                "P5",
                 "Q25",
                 "中位数",
                 "均值",
                 "Q75",
+                "P95",
                 "最大值",
                 "标准差",
+                "CV%",
+                "偏度",
+                "峰度",
             ]
             for col, s in numeric_summary.items():
                 stats_table["rows"].append(
                     [
                         col,
                         round(s["min"], 2),
+                        round(s["p5"], 2),
                         round(s["q25"], 2),
                         round(s["median"], 2),
                         round(s["mean"], 2),
                         round(s["q75"], 2),
+                        round(s["p95"], 2),
                         round(s["max"], 2),
                         round(s["std"], 2),
+                        s["cv"],
+                        s["skewness"],
+                        s["kurtosis"],
                     ]
                 )
 
@@ -195,12 +698,25 @@ def analyze_csv(file_path):
         # ==========================================
         chart_data = {
             "overview": overview,
+            "data_quality": data_quality,
             "numeric_cols": numeric_cols,
             "distributions": distributions,
             "correlations": correlations,
+            "correlation_highlights": correlation_highlights,
             "categories": categories,
+            "segment_breakdown": segment_breakdown,
             "time_series": time_series,
+            "time_series_multi": time_series_multi,
+            "time_series_diagnostics": time_series_diagnostics,
             "scatter": scatter,
+            "box_plots": box_plots,
+            "outliers": outliers,
+            "primary_metric": primary_metric,
+            "anomaly_overview": anomaly_overview,
+            "driver_analysis": driver_analysis,
+            "segment_comparison": segment_comparison,
+            "top_bottom": top_bottom,
+            "ranking_signal": ranking_signal,
             "stats_table": stats_table,
         }
 
@@ -214,21 +730,125 @@ def analyze_csv(file_path):
             "【数据概览】",
             f"- 数据集尺寸: {overview['rows']} 行 × {overview['cols']} 列",
             f"- 缺失值情况: 共有 {overview['missing_cells']} 个单元格缺失，整体数据完整率 {100 - overview['missing_pct']}%",
+            f"- 重复行: {overview['duplicate_rows']} 行",
+            f"- 内存占用: {overview['memory_kb']} KB",
             f"- 数值型列 ({len(numeric_cols)}): {', '.join(numeric_cols[:10])}",
             f"- 分类型列 ({len(categorical_cols)}): {', '.join(categorical_cols[:10])}",
+            f"- 数据类型分布: {dtype_counts}",
             "",
-            "【数值型特征统计 (Top 8)】",
+            "【质量关注点】",
         ]
+        quality_focus = [
+            (col, rate, miss_count)
+            for col, rate, miss_count in missing_by_col[:5]
+            if rate > 0
+        ]
+        if quality_focus:
+            for col, rate, miss_count in quality_focus:
+                summary_lines.append(f"- {col}: 缺失 {miss_count} 个，占比 {rate}%")
+        else:
+            summary_lines.append("- 所有字段均无缺失，数据完整性较高")
+
+        summary_lines.extend(
+            [
+                "",
+                "【数值型特征统计 (Top 8)】",
+            ]
+        )
         for col, s in numeric_summary.items():
             summary_lines.append(
-                f"- {col}: min={s['min']:.2f}, Q25={s['q25']:.2f}, median={s['median']:.2f}, "
-                f"mean={s['mean']:.2f}, Q75={s['q75']:.2f}, max={s['max']:.2f}, std={s['std']:.2f}"
+                f"- {col}: min={s['min']:.2f}, P5={s['p5']:.2f}, Q25={s['q25']:.2f}, "
+                f"median={s['median']:.2f}, mean={s['mean']:.2f}, Q75={s['q75']:.2f}, "
+                f"P95={s['p95']:.2f}, max={s['max']:.2f}, std={s['std']:.2f}, "
+                f"CV={s['cv']}%({classify_cv(s['cv'])}), spread={s['spread']:.2f}, "
+                f"skew={s['skewness']}({classify_skewness(s['skewness'])}), kurtosis={s['kurtosis']}"
             )
 
+        if numeric_summary:
+            volatile_cols = sorted(
+                numeric_summary.items(), key=lambda x: x[1]["cv"], reverse=True
+            )[:3]
+            summary_lines.append("")
+            summary_lines.append("【波动性与偏态重点】")
+            for col, s in volatile_cols:
+                summary_lines.append(
+                    f"- {col}: 波动等级={classify_cv(s['cv'])}, CV={s['cv']}%, 偏态={classify_skewness(s['skewness'])}"
+                )
+
+        # 异常值摘要
+        if outliers:
+            summary_lines.append("")
+            summary_lines.append("【异常值检测 (IQR 方法)】")
+            for col, info in outliers.items():
+                if info["count"] > 0:
+                    summary_lines.append(
+                        f"- {col}: {info['count']} 个异常值 ({info['pct']}%), "
+                        f"正常范围 [{info['lower_bound']}, {info['upper_bound']}]"
+                    )
+            if not any(info["count"] > 0 for info in outliers.values()):
+                summary_lines.append("- 未检测到显著异常值")
+
+        # Top/Bottom 排名
+        if top_bottom:
+            summary_lines.append("")
+            summary_lines.append(
+                f"【Top 5 / Bottom 5 排名 (按 {top_bottom['rank_col']})】"
+            )
+            summary_lines.append(
+                f"  Top 5: {list(zip(top_bottom['top5']['labels'], top_bottom['top5']['values']))}"
+            )
+            summary_lines.append(
+                f"  Bottom 5: {list(zip(top_bottom['bottom5']['labels'], top_bottom['bottom5']['values']))}"
+            )
+            if ranking_signal:
+                summary_lines.append(
+                    f"  排名断层: Top5均值={ranking_signal['top_avg']}, Bottom5均值={ranking_signal['bottom_avg']}, 差值={ranking_signal['gap']}"
+                )
+
+        if anomaly_overview:
+            summary_lines.append("")
+            summary_lines.append("【数据异动概述】")
+            summary_lines.append(
+                f"- 核心分析指标: {anomaly_overview['metric']}，均值={anomaly_overview['mean']}，中位数={anomaly_overview['median']}，P10={anomaly_overview['q10']}，P90={anomaly_overview['q90']}"
+            )
+            summary_lines.append(
+                f"- 高位组({anomaly_overview['top_group_size']}个样本)均值={anomaly_overview['top_group_mean']}，低位组({anomaly_overview['bottom_group_size']}个样本)均值={anomaly_overview['bottom_group_mean']}，差值={anomaly_overview['gap']}"
+            )
+            summary_lines.append(
+                f"- 主指标异常值数量={anomaly_overview['outlier_count']}，占比={anomaly_overview['outlier_pct']}%，分位带样本分布={list(zip(anomaly_overview['band_labels'], anomaly_overview['band_values']))}"
+            )
+
+        if driver_analysis.get("items"):
+            summary_lines.append("")
+            summary_lines.append("【归因分析线索】")
+            for item in driver_analysis["items"][:5]:
+                summary_lines.append(
+                    f"- {item['name']}: 综合驱动分={item['score']}，与 {driver_analysis['metric']} 的相关系数={item['corr']}，高位组均值={item['top_mean']}，低位组均值={item['bottom_mean']}，组间差异强度={item['gap_ratio']}"
+                )
+
         summary_lines.append("")
         summary_lines.append("【分类型特征摘要 (Top 6)】")
         for col, stats in cat_summary.items():
-            summary_lines.append(f"- {col}: {stats}")
+            summary_lines.append(
+                f"- {col}: 唯一值={stats['n_unique']}, 最常见={stats['top1']} "
+                f"(出现{stats['top1_count']}次, 占比{stats['top1_share']}%), 熵={stats['entropy']}, "
+                f"Top3集中度={stats['top3_share']}%"
+            )
+
+        if segment_breakdown:
+            summary_lines.append("")
+            summary_lines.append("【分类维度切片表现】")
+            for segment in segment_breakdown:
+                leaders = segment["leaders"][:3]
+                leader_text = "; ".join(
+                    [
+                        f"{item['name']}(样本{item['count']}, 均值{item['mean']}, 总量{item['sum']})"
+                        for item in leaders
+                    ]
+                )
+                summary_lines.append(
+                    f"- 维度 {segment['dimension']} 对指标 {segment['metric']} 的高贡献分组: {leader_text}"
+                )
 
         summary_lines.append("")
         summary_lines.append("【核心相关性】")
@@ -244,6 +864,14 @@ def analyze_csv(file_path):
                 summary_lines.extend([f"- {c}" for c in strong_corrs])
             else:
                 summary_lines.append("- 没有发现强相关的数值变量组合（|r| >= 0.5）。")
+            if correlation_highlights["positive"]:
+                summary_lines.append("- 最高正相关组合:")
+                for c1, c2, val in correlation_highlights["positive"]:
+                    summary_lines.append(f"  * {c1} vs {c2}: {val}")
+            if correlation_highlights["negative"]:
+                summary_lines.append("- 最低相关组合:")
+                for c1, c2, val in correlation_highlights["negative"]:
+                    summary_lines.append(f"  * {c1} vs {c2}: {val}")
 
         if scatter:
             summary_lines.append("")
@@ -251,12 +879,33 @@ def analyze_csv(file_path):
                 f"【散点图】已生成 {scatter['x_name']} vs {scatter['y_name']} 的散点图数据"
             )
 
+        # 时间序列摘要
+        if time_series["dates"]:
+            summary_lines.append("")
+            summary_lines.append(
+                f"【时间序列】检测到时间列，已按月/日聚合 {time_series['name']} 趋势"
+            )
+            if time_series_diagnostics:
+                summary_lines.append(
+                    f"- 时间字段={time_series_diagnostics['date_col']}, 观测点={time_series_diagnostics['points']}, "
+                    f"起点={time_series_diagnostics['start']}, 终点={time_series_diagnostics['end']}, "
+                    f"整体变化={time_series_diagnostics['change_pct']}%, 斜率={time_series_diagnostics['slope']}, "
+                    f"波动率={time_series_diagnostics['volatility_pct']}%"
+                )
+                summary_lines.append(
+                    f"- 峰值出现在 {time_series_diagnostics['peak_date']} ({time_series_diagnostics['peak_value']}), "
+                    f"谷值出现在 {time_series_diagnostics['trough_date']} ({time_series_diagnostics['trough_value']})"
+                )
+            if time_series_multi:
+                extra_names = [ts_m["name"] for ts_m in time_series_multi]
+                summary_lines.append(f"  额外趋势列: {', '.join(extra_names)}")
+
         summary_lines.append("==================================================")
         summary_lines.append(
-            "请作为数据分析专家，基于以上【统计摘要】为用户撰写深度的数据分析见解（Insights）。"
+            "请作为数据分析专家，基于以上【统计摘要】为用户撰写深度的数据分析见解（Insights）。每个模块尽量覆盖现象、可能原因、业务影响、行动建议四层内容，避免只重复统计值。"
         )
         summary_lines.append(
-            "并且，在使用 html_interpreter 时，请将下方 JSON_START 和 JSON_END 标记之间的纯 JSON 字符串完整传递给变量 CHART_DATA_JSON。"
+            "注意：marker 中包裹的 CHART_DATA_JSON 会由后端自动注入模板，你无需手动传递。"
         )
         summary_lines.append("###CHART_DATA_JSON_START###")
         summary_lines.append(chart_data_json_str)