数据工具箱(专业版)

面向数据团队的全功能数据工程平台，支持统计检验、质量监控、工作流自动化、高级可视化和数据血缘追踪管理。

功能说明: 本技能涵盖 化工作流场景 等核心能力。

概述

专业版在免费版基础数据处理能力上，扩展为面向数据团队的全功能工程平台。覆盖统计检验、数据质量监控、工作流自动化、高级可视化、数据血缘追踪与多源连接优化，适合企业级数据管道、质量治理与 A/B 实验分析场景. 专业版将数据处理从"手动单次操作"升级为"自动化可追溯工作流"，新增数据质量评分与漂移检测引擎，提供完整统计检验体系支撑数据驱动决策，并通过血缘追踪实现转换谱系的端到端可审计.

核心能力

能力域	说明	专业版独有
基础处理	提取/清洗/分析/基础可视化	否（免费版可用）
统计检验	t检验/卡方/ANOVA/非参数	是
质量监控	质量评分/漂移检测/异常告警	是
工作流自动化	DAG编排/定时/检查点/恢复	是
高级可视化	热力图/桑基图/交互式图表	是
数据血缘	转换谱系/影响分析/版本对比	是
多源优化	跨源JOIN/查询路由/缓存	是
技术实现要点：核心能力基于input_params参数与output_format配置实现,支持创建/查询/修改/删除等操作模式,通过config_options进行运行时配置.

核心功能执行

用input_params参数进行配置.

处理: 解析核心功能执行的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回核心功能执行的响应数据,包含状态码、结果和日志.

• 执行此能力时使用input_params参数,支持创建/查询/导出操作

参数配置与调用

用config_options参数进行配置.

处理: 解析参数配置与调用的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回参数配置与调用的响应数据,包含状态码、结果和日志.

• 执行此能力时使用config_options参数,支持修改/重置/导入操作

结果处理与输出

用output_format参数进行配置.

处理: 解析结果处理与输出的输入参数,完成核心逻辑,返回结构化响应. 输出: 返回结果处理与输出的响应数据,包含状态码、结果和日志.

• 执行此能力时使用output_format参数,支持导出/保存/转换操作 能力覆盖范围：本skill的核心能力覆盖以下场景关键词：全功能数据工程平、覆盖统计检验、工作流自动化与高、数据工具箱专业版、是一款面向数据团、队的全功能数据工、程平台、在免费版基础数据、处理能力上扩展统、数据质量监控、数据血缘追踪与多、源连接优化等能力、Use、when、需要数据分析、报表生成、统计洞察、数据可视化时使用、不适用于实时流数、据处理、适用于独立开发者、企业团队和自动化、工作流场景等。这些关键词对应description中声明的使用场景,均已在上述能力点中提供对应的操作支持.

使用场景

场景一：A/B 实验统计决策（研究员/产品）

产品团队上线新功能后需判断实验组与对照组的转化率差异是否显著。专业版提供完整统计检验流程：

python

from scipy import stats
# ...
# A/B 实验数据
control = [0.12, 0.13, 0.11, 0.14, 0.12]  # 对照组每日转化率
treatment = [0.15, 0.16, 0.14, 0.17, 0.15]  # 实验组每日转化率
# ...
# 正态性检验
_, p_normal = stats.shapiro(control + treatment)
if p_normal > 0.05:
    # 正态分布：使用 t 检验
    t_stat, p_value = stats.ttest_ind(treatment, control)
    test_name = "独立样本 t 检验"
else:
    # 非正态：使用 Mann-Whitney U 检验
mannwhitneyu(treatment, control)
    test_name = "Mann-Whitney U 检验"
# ...
print(f"检验方法: {test_name}")
print(f"统计量: {t_stat:.4f}, p 值: {p_value:.4f}")
print(f"结论: {'显著' if p_value < 0.05 else '不显著'}（α=0.05）")

场景二：数据质量监控与漂移检测（工程师/运维）

生产环境数据质量需持续监控。专业版提供质量评分与漂移检测引擎：

python

# 数据质量评分
quality_report = {
  'completeness': 1 - (df.isnull().sum().sum() / df.size),  # 完整性
  'uniqueness': df['id'].nunique() / len(df),                # 唯一性
  'validity': (df['amount'] >= 0).mean(),                    # 有效性
  'consistency': check_consistency(df),                      # 一致性
}
overall_score = sum(quality_report.values()) / len(quality_report)
# ...
# 漂移检测：对比本周与上周数据分布
from scipy.stats import ks_2samp
_, p_drift = ks_2samp(this_week['amount'], last_week['amount'])
if p_drift < 0.05:
    alert("数据漂移告警：amount 字段分布发生显著变化")

场景三：自动化工作流编排（工程师）

数据管道需按 DAG 编排，支持检查点与故障恢复：

python

# 工作流定义（DAG）
workflow = {
  'name': '每日收入报表',
  'schedule': '0 8 * * *',
  'tasks': [
    {'id': 'extract', 'script': 'extract.py', 'depends_on': []},
    {'id': 'clean', 'script': 'clean.py', 'depends_on': ['extract']},
    {'id': 'analyze', 'script': 'analyze.py', 'depends_on': ['clean']},
    {'id': 'report', 'script': 'report.py', 'depends_on': ['analyze']},
  ],
  'checkpoint': True,
  'retry': {'max': 3, 'backoff': 'exponential'}
}

场景四：数据血缘追踪（合规/审计）

审计需追溯每个报表字段的来源与转换链路：

bash

# 查询字段血缘
uv run lineage.py --trace field 'report.revenue' --output tree

输出：

text

report.revenue
├── clean.amount (清洗: 去除空值)
│   ├── raw.amount (源: stripe.transactions)
│   └── raw.fee (源: stripe.fees, 转换: amount - fee)
└── clean.currency (清洗: 统一为 CNY)
    └── raw.currency (源: stripe.transactions, 转换: 汇率换算)

快速开始

前置条件

• Python 3.11 及以上版本
• 已安装 pandas、scipy、matplotlib
• 数据源访问凭据（数据库连接串、API Key 等）

120 秒上手

领先步，安装依赖：

bash

pip install pandas scipy matplotlib plotly

第二步，向 Agent 描述任务：

text

帮我对这份数据做完整的质量评估，包括完整性、唯一性、有效性评分

第三步，配置工作流：

bash

uv run workflow.py --init --config daily_report.json
uv run workflow.py --schedule "0 8 * * *"

响应解析: 完成完成后,查看输出响应确认任务状态。成功时输出包含解析摘要和响应数据;失败时根据错误信息排查问题,查阅错误解析章节获取恢复步骤.

示例

统计检验配置

检验场景	推荐方法	前提条件
两组均值对比	独立样本 t 检验	正态分布、方差齐性
两组非正态对比	Mann-Whitney U	无分布假设
多组均值对比	ANOVA	正态分布、方差齐性
分类变量关联	卡方检验	期望频数 ≥5
配对样本对比	配对 t 检验 / Wilcoxon	差值正态/非正态

数据质量评分配置

json

{
  "quality_dimensions": {
    "completeness": {"weight": 0.3, "threshold": 0.95},
    "uniqueness": {"weight": 0.2, "threshold": 0.99},
    "validity": {"weight": 0.3, "threshold": 0.97},
    "consistency": {"weight": 0.2, "threshold": 0.95}
  },
  "drift_detection": {
    "method": "ks_test",
    "alpha": 0.05,
    "baseline_window": "7d",
    "compare_window": "1d"
  },
  "alert": {
    "on_quality_drop": true,
    "on_drift": true,
    "webhook": "https://your-hook.example/notify"
  }
}

工作流 DAG 配置

json

{
  "name": "每日数据管道",
  "schedule": "0 6 * * *",
  "tasks": [
    {"id": "extract_stripe", "script": "extract_stripe.py", "depends_on": []},
    {"id": "extract_db", "script": "extract_db.py", "depends_on": []},
    {"id": "merge", "script": "merge.py", "depends_on": ["extract_stripe", "extract_db"]},
    {"id": "clean", "script": "clean.py", "depends_on": ["merge"]},
    {"id": "analyze", "script": "analyze.py", "depends_on": ["clean"]},
    {"id": "visualize", "script": "visualize.py", "depends_on": ["analyze"]},
    {"id": "report", "script": "report.py", "depends_on": ["visualize"]}
  ],
  "checkpoint": true,
  "retry": {"max": 3, "backoff": "exponential", "base": 60},
  "on_failure": {"notify": "webhook", "retry_from_checkpoint": true}
}

高级可视化

python

import plotly.express as px
import plotly.graph_objects as go
# ...
# 交互式折线图
fig = px.line(df, x='date', y='revenue', title='收入趋势（可缩放）')
fig.write_html('revenue_interactive.html')
# ...
# 热力图：相关性矩阵
corr = df.corr()
fig = go.Figure(data=go.Heatmap(z=corr.values, x=corr.columns, y=corr.columns))
fig.write_html('correlation_heatmap.html')
# ...
# 桑基图：用户流向
fig = go.Figure(data=[go.Sankey(
  node={'label': ['访问', '注册', '试用', '付费', '流失']},
  link={'source': [0,1,2,0,1], 'target': [1,2,3,4,4], 'value': [500,300,100,200,100]}
)])
fig.write_html('user_flow_sankey.html')

优选实践

1. 统计检验前置假设验证

执行统计检验前必须验证前提假设：正态性（Shapiro-Wilk）、方差齐性（Levene）。假设不满足时切换非参数检验，避免结论失真.

2. 数据质量持续监控

将质量评分纳入数据管道的每日检查，质量下降时自动告警。漂移检测使用 KS 检验对比当前与基线分布，p 值低于 0.05 时触发告警.

3. 工作流检查点与幂等

每个任务完成后写入检查点，故障恢复时从最后成功的检查点继续，避免从头重放。任务设计为幂等，重复执行不产生副作用.

4. 血缘追踪自动化

在清洗与转换脚本中自动记录输入字段、操作与输出字段的映射关系，生成血缘图谱。新增报表字段时自动追溯其全部来源.

5. 多源查询路由

跨数据源查询时根据数据量与延迟选择最优执行策略：小表本地 JOIN、大表下推至数据库、频繁查询结果缓存.

常见问题

Q1：t 检验和 Mann-Whitney U 怎么选？

先对数据做 Shapiro-Wilk 正态性检验。p > 0.05 且方差齐性满足时用 t 检验；否则用 Mann-Whitney U 非参数检验。样本量大于 30 时 t 检验对非正态有一定鲁棒性（中心极限定理）.

Q2：数据漂移检测用什么方法？

连续变量使用 KS 检验（Kolmogorov-Smirnov）对比分布；分类变量使用卡方检验对比分布；时间序列使用 ADWIN 或 Page-Hinkley 检测概念漂移.

Q3：工作流故障后如何恢复？

启用检查点模式后，故障恢复从最后成功的任务继续执行。若任务非幂等，需手动确认后恢复。配置 retry_from_checkpoint: true 实现自动恢复.

Q4：血缘追踪如何实现？

在转换脚本中调用 lineage.record() 记录输入输出字段映射。系统自动构建血缘图谱，支持正向追踪（字段影响哪些下游）与反向追溯（字段来自哪些上游）.

Q5：多源 JOIN 性能差？

避免跨源全量 JOIN。策略：小表（<1GB）拉取到本地 JOIN；大表在数据库端完成 JOIN 后拉取结果；频繁查询的 JOIN 结果缓存 5 分钟.

Q6：数据库 查询如何优化？

使用 EXPLAIN ANALYZE 查看执行计划。常见优化：为 JOIN/WHERE 字段添加索引、使用分区表加速时间范围查询、避免子查询嵌套过深、使用 CTE 提升可读性与性能.

Q7：质量评分如何设定阈值？

参考行业基准与业务需求。一般完整性 ≥95%、唯一性 ≥99%、有效性 ≥97%。关键业务字段（如金额、用户ID）阈值应更高。评分低于阈值时告警但不阻塞管道.

Q8：交互式图表如何分享？

Plotly 生成的 HTML 文件自带渲染引擎，可直接分享给他人用浏览器打开。若需嵌入内部系统，使用 Plotly 的 iframe 嵌入模式.

Q9：如何管理多个数据管道？

使用工作流注册表统一管理。每个管道记录名称、调度、依赖、状态。支持按优先级排队、资源隔离与依赖编排.

Q10：如何做可复现的数据分析？

固定随机种子（np.random.seed(42)）、记录包版本（pip freeze > requirements.txt）、使用 Jupyter Notebook 的检查点、将数据快照与代码一并归档.

依赖说明

运行环境

• Agent 平台：支持 SKILL.md 的任意 AI Agent（Claude Code / Cursor / Codex / Gemini CLI 等）
• 操作系统：Windows / macOS / Linux
• Python：3.11 及以上版本

依赖详情

依赖项	类型	是否必需	获取方式	版本兼容性
pandas	Python 库	必需	pip install pandas	2.0+
scipy	Python 库	统计检验必需	pip install scipy	1.10+
matplotlib	Python 库	基础可视化必需	pip install matplotlib	3.7+
plotly	Python 库	高级可视化必需	pip install plotly	5.18+
psql	数据库客户端	数据库源推荐	apt install 数据库-client	12+
LLM API	API	必需	由 Agent 平台内置 LLM 提供	不限

API Key 配置

• 数据库连接串通过环境变量 DATABASE_URL 注入
• API 数据源 Token 存储于环境变量
• 告警 Webhook 地址存储于环境变量 DATA_ALERT_WEBHOOK
• 工作流调度凭据存储于 $HOME/.data-toolkit/credentials/（已 gitignore）
• 禁止在脚本或配置文件中硬编码凭据

可用性分类

• 分类：MD+EXEC（纯 Markdown 指令，核心功能需 exec 命令行执行能力）
• 说明：基于 Markdown 的 AI Skill，通过自然语言指令驱动 Agent 执行数据工程任务

专业版特性

本专业版相比免费版新增以下能力：

• 统计检验体系：t 检验/卡方/ANOVA/非参数检验，含假设验证流程
• 数据质量监控：四维质量评分、漂移检测、异常告警
• 工作流自动化：DAG 编排、定时调度、检查点与故障恢复
• 高级可视化：热力图/桑基图/交互式图表/仪表盘
• 数据血缘追踪：转换谱系记录、影响分析、版本对比
• 多源连接优化：跨源 JOIN、查询路由、缓存策略
• 可复现性保障：随机种子、版本锁定、数据快照归档
• 优先技术支持：工作日 4 小时内响应，提供 SLA 保障

定价

版本	价格	功能	适用场景
免费体验版	0 元	基础处理 + 基础可视化	个人试用
收费专业版	199 元/月	全功能 + 统计检验 + 工作流 + 优先支持	团队/企业

专业版通过 Skill 平台付费发布，支持按月订阅与一次性买断（1999 元）.

错误处理

错误场景	原因	处理方式
配置错误	参数缺失或格式错误	检查依赖说明中的配置要求
运行时错误	运行环境不满足	确认运行环境符合依赖说明
网络错误	连接超时或不可达	执行ping命令测试网络连通性,检查防火墙和代理设置连接后执行ping命令测试网络连通性,检查防火墙和代理设置连接后重新执行命令，参考国内替代方案

已知限制

• 需LLM支持,无LLM环境不可用
• 复杂业务场景建议结合人工经验判断
• 执行效率受模型能力与网络环境影响

安全注意事项

风险类型	防范措施
API密钥泄露	通过环境变量配置，禁止硬编码到代码或配置文件中
命令执行风险	仅执行白名单命令，避免拼接用户输入到命令行参数中
网络通信安全	使用HTTPS协议，验证SSL证书有效性
敏感数据暴露	输出结果中不包含密钥、令牌等敏感信息

使用前请确认已阅读依赖说明章节，确保运行环境满足安全要求。