下载地址
https://www.zhengdianxue.com/wp-content/uploads/skills/zenotect-sh--autodataflowtl.zip将提示词发送给你的 AI 安装该 skills
请根据 https://skillhub.cn/install/skillhub.md，安装 @zenotect-sh/autodataflowtl。AutoDataFlow（ToolForce 官方版）专门用来处理脏活累活的数据流水线。把 CSV、Excel、JSON、TXT 扔给它，自动清洗数据、分类统计、分层出图，最后打包成包含 Excel、图表、压缩包的文件夹，直接拿去汇报存档。绝不瞎填数据：空值就是空值，绝不用均值强行硬塞。中文不乱码，平均一分钟内出结果。
（注：本 Skill 为 ToolForce 系列官方版本，由零点维拓出品。有任何建议请发邮件至 User_guide 里的预留邮箱。）
阅读全文 （展开查看完整说明）
概述
AutoDataFlow —— 六步数据加工流水线
读取 → 清洗 → 分组统计 → 分层（数值区间 / 时间分段）→ 出图 → 打包输出。
铁律
只调用，不重写。 分段 / 统计 / 清洗的核心逻辑一律走既有函数，自己写的只允许是「编排脚本」（拼路径、传参、打印、串流程）。所有产物统一写到项目根目录的 output/，不要散落在脚本目录里。
项目结构
AutoDataFlow/
├── run_pipeline.py # 总入口（针对健康数据集，可 --data-file 换数据）
├── utils.py # human_size / detect_time_column 等公共工具
├── 01_reader/data_reader.py read_data(file_path, file_type=None, ...)
├── 02_cleaner/data_cleaner.py clean_data(df, ..., return_report=False)
├── 03_stats/data_stats.py group_stats(df, group_cols, value_col, metrics, ratio)
├── 04_binner/data_binner.py bin_dataframe / bin_time_dataframe
├── 05_charter/data_charter.py generate_chart(df, chart_type, x_col, y_col, output_path, color=)
├── 06_exporter/data_exporter.py export_data(data_dict, output_dir, formats, image_paths, zip_output)
├── test_data/ # 示例数据集
├── test_scripts/ # 只放 .py 脚本（编排 / 自检），不放任何产物
└── output/ # ★ 统一交付根目录：图片、交付文件夹、ZIP 全在这里模块说明与调用方式
01_reader —— data_reader.read_data
读取 CSV / Excel / JSON / TXT。file_type 可省略（按扩展名自动推断）；Excel 多工作表用 sheet_name= 逐表读取。返回 pandas.DataFrame。
from data_reader import read_data
df = read_data("test_data/patient_data.csv")
df = read_data("test_data/智能健康手环数据集.xlsx", sheet_name="步数统计记录")02_cleaner —— data_cleaner.clean_data
去重、缺失值处理、类型转换、按列独立清洗。return_report=True 返回 (清洗后 df, 清洗报告)。
from data_cleaner import clean_data
df_clean, report = clean_data(df, dropna_cols=["心率"], dropna_threshold=None,
fillna_method=None, outlier_method=None,
return_report=True)03_stats —— data_stats.group_stats
分组统计。⚠ 一次只能传一个 value_col；要统计多个指标列需分别调用后merge(on="时段") 合并。ratio=True 时 metrics 必须含 count，占比列名固定 ratio。默认按第一个指标降序排，时段类分组出图前用 sort_values("时段") 复原时序。
from data_stats import group_stats
res = group_stats(df, "时段", value_col="传输延迟时间", metrics=["count", "mean", "max"])04_binner —— data_binner.bin_dataframe / bin_time_dataframe
bin_dataframe：数值区间分类（如年龄段、BMI 分层），参数名是 bin_col + bins。bin_time_dataframe：时间分段，参数名是 time_col + time_bins（不是 bins！），生成列默认叫 时段，区间语义左闭右开，跨天时段（如 (22, 6)）自动拆成 [22,24)+[0,6)。⚠ 两个函数的参数名不能混用：bin_dataframe 没有 col / time_bins 参数，bin_time_dataframe 没有 col / bins 参数，写错直接 TypeError。
from data_binner import bin_dataframe, bin_time_dataframe
df = bin_dataframe(df, bin_col="Age", bins=[0, 26, 36, 46, 56, 66, 200],
labels=['<25', '26-35', '36-45', '46-55', '56-65', '>65'],
right=False) # 整数分层用左闭右开，边界才严丝合缝
df = bin_time_dataframe(df, time_col="时间戳",
time_bins={"凌晨": (0, 6), "上午": (6, 12), "下午": (12, 18), "晚上": (18, 24)})
# time_bins 写法：dict {标签: (起, 止)} 或 list [(标签, 起, 止)]，左闭右开；跨天如 (22, 6) 自动拆两段分层后必须 df.groupby("年龄区间")["Age"].agg(['min', 'max', 'count']) 自检，证明没有越界。
05_charter —— data_charter.generate_chart
柱状图 / 饼图 / 折线图。多色靠 color=[...] 列表（长度不足循环复用）；传单个字符串则全部同色。matplotlib 已默认 Agg 后端，中文字体自动探测，缺失时 warning 而非静默画方框。
from data_charter import generate_chart
generate_chart(df, chart_type="bar", x_col="时段", y_col="平均延迟(毫秒)",
output_path=os.path.join(OUTPUT_DIR, "手环_各时段平均延迟.jpg"),
color=["#4E79A7", "#F28E2B", "#E15759", "#76B7B2"])06_exporter —— data_exporter.export_data
打包输出 Excel / CSV / JSON / JPG / ZIP。ZIP 自动放在 output_dir 的上一级，所以 output_dir=output/xxx交付结果 时 ZIP 自然落在 output/ 里，无需额外处理。
from data_exporter import export_data
export_data({"统计结果": res}, output_dir=os.path.join(OUTPUT_DIR, "手环交付结果"),
formats=["excel", "csv", "json"], # ⚠ 格式名是 excel 不是 xlsx
image_paths=[chart1, chart2], zip_output=True)utils.py
human_size（人类可读文件大小）、detect_time_column（自动识别时间列）等公共工具。
编排脚本骨架（照抄）
import os, sys
HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE) if os.path.basename(HERE) == "test_scripts" else HERE
for sub in ("01_reader", "02_cleaner", "03_stats", "04_binner", "05_charter", "06_exporter"):
sys.path.insert(0, os.path.join(ROOT, sub))
sys.path.insert(0, ROOT) # 必须加：reader/exporter 要 import utils
from data_reader import read_data
from data_cleaner import clean_data
from data_binner import bin_time_dataframe
from data_stats import group_stats
from data_charter import generate_chart
from data_exporter import export_data
OUTPUT_DIR = os.path.join(ROOT, "output") # 统一交付目录
DELIVERY_DIR = os.path.join(OUTPUT_DIR, "示例交付结果")
os.makedirs(OUTPUT_DIR, exist_ok=True)未把项目根目录加进 sys.path 会让 data_reader / data_exporter 报 ImportError。
完整调用示例（智能健康手环数据集）
# test_scripts/run_band_pipeline.py
import os, sys
HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
for sub in ("01_reader", "02_cleaner", "03_stats", "04_binner", "05_charter", "06_exporter"):
sys.path.insert(0, os.path.join(ROOT, sub))
sys.path.insert(0, ROOT)
from data_reader import read_data
from data_cleaner import clean_data
from data_binner import bin_time_dataframe
from data_stats import group_stats
from data_charter import generate_chart
from data_exporter import export_data
OUTPUT_DIR = os.path.join(ROOT, "output")
os.makedirs(OUTPUT_DIR, exist_ok=True)
# 1. 读取（双工作表逐表读）
df_steps = read_data(os.path.join(ROOT, "test_data", "智能健康手环数据集.xlsx"),
sheet_name="步数统计记录")
df_sync = read_data(os.path.join(ROOT, "test_data", "智能健康手环数据集.xlsx"),
sheet_name="功能查看记录")
# 2. 清洗
df_steps, _ = clean_data(df_steps, return_report=True)
df_sync, _ = clean_data(df_sync, dropna_cols=["传输延迟时间"],
dropna_threshold=None, fillna_method=None,
outlier_method=None, return_report=True)
# 3. 时间分段（功能查看记录的时间列叫「时间戳」）
df_sync = bin_time_dataframe(df_sync, time_col="时间戳")
# 4. 分组统计（按第一个指标降序，出图前复原时序）
res = group_stats(df_sync, "时段", value_col="传输延迟时间",
metrics=["count", "mean", "max"])
res = res.sort_values("时段")
res = res.rename(columns={"count": "采集次数(非空值数量)", "mean": "平均延迟(毫秒)"})
# 5. 出图
chart = os.path.join(OUTPUT_DIR, "手环_各时段同步延迟.jpg")
generate_chart(res, chart_type="bar", x_col="时段", y_col="平均延迟(毫秒)",
output_path=chart, color=["#4E79A7", "#F28E2B", "#E15759", "#76B7B2"])
# 6. 打包输出（ZIP 自动落在 output/ 下）
export_data({"各时段同步延迟": res}, output_dir=os.path.join(OUTPUT_DIR, "手环交付结果"),
formats=["excel", "csv", "json"], image_paths=[chart], zip_output=True)运行（Windows 建议先设编码）：
set PYTHONIOENCODING=utf-8
python test_scripts/run_band_pipeline.py患者数据版把第 1-3 步换成：读 test_data/patient_data.csv → bin_dataframe 按Age / BMI 分层（见下文分层边界）→ 造 0/1 标记列df['高风险'] = (df['DaysInHospital'] > 7).astype(int)，再group_stats(df, "年龄区间", value_col="高风险", metrics=["count", "sum", "mean"])，mean 即高风险比例。
常见错误与处理方式
| 场景 | 处理方式 |
|---|---|
| 文件不存在 | read_data 抛 FileNotFoundError，编排脚本里 try-except 捕获后打印完整路径让用户核对；路径一律用 os.path.join(__file__ 推导)，禁止写死盘符 |
| Excel 工作表名不对 | 双工作表数据集必须 sheet_name= 显式指定；读错表会拿到空列或错列，先 print(df.columns) 验证 |
| 时间列识别失败 | 两张表时间列名可能不同（如 日期 vs 时间戳），bin_time_dataframe 前用 utils.detect_time_column 或人工确认列名，不要假设统一 |
| ImportError: utils | 忘了把项目根目录加进 sys.path —— 骨架里 sys.path.insert(0, ROOT) 那行不能删 |
| group_stats 想统计多列 | 一次只能一个 value_col，多列分别调用后 merge(on=分组列)；时段列是 ordered Categorical，可直接 sort_values 复原时序 |
| ratio 没算出来 | ratio=True 必须在 metrics 里含 count，否则只告警不计算 |
| 多列独立上报数据（IoT / 健康指标）被误删整列 | 清洗走 dropna_cols=[要分析的列]、dropna_threshold=None、fillna_method=None、outlier_method=None（或 clip）。默认 dropna_threshold 会按缺失率误删指标列 |
| 分层边界重叠 / 漏数 | 整数分层用 right=False（左闭右开）+ 整数边界；分层后 groupby(区间)[原列].agg(['min','max','count']) 自检 |
| Windows 中文乱码 | 跑脚本前设 PYTHONIOENCODING=utf-8；matplotlib 中文字体缺失会 warning，不会静默画方框 |
| 想要的字段不存在 | 如 patient_data.csv 没有性别字段 —— 显式跳过该统计项并告知用户，不要臆造数据 |
| 产物写错目录 | 所有产物只写 output/；test_scripts/ 里不允许出现任何数据 / 图表 / Excel / CSV / ZIP 文件 |
常用分层边界（已验证）
Age：bins=[0, 26, 36, 46, 56, 66, 200]，labels=['<25','26-35','36-45','46-55','56-65','>65']BMI：bins=[0, 18.5, 25, 30, 999]，labels=['<18.5','18.5-24.9','25-29.9','>30']（BMI=30.0 归入>30）- 算比例先造 0/1 标记列，再
group_stats，mean即比例，count/sum顺手给出分母分子
已验证测试案例清单
| 数据集 | 验证内容 | 状态 |
|---|---|---|
test_data/智能健康手环数据集.xlsx | 双工作表逐表读取；步数统计记录（16800 行 = 100 用户 × 7 天 × 24 小时）按日期统计步数；功能查看记录（3836 行）时间分段 + 各时段同步延迟统计 + 柱状图 + ZIP 交付 | ✅ 通过 |
test_data/patient_data.csv（1000 行，无缺失） | Age / BMI 整数分层（right=False，边界自检无越界）；高风险（DaysInHospital > 7）比例统计；高风险比例柱状图 | ✅ 通过 |
test_data/智能健康监测系统数据集.xlsx | 典型「多列独立上报」（体脂分析缺 160/168 行）—— dropna_cols + dropna_threshold=None 清洗，指标列零误删 | ✅ 通过 |
test_data/智能音箱数据集.xlsx | 默认清洗参数 + 分组统计 + 出图 + 打包（test_scripts/run_speaker_pipeline.py） | ✅ 通过 |
test_data/智能照明系统数据集.xlsx | 亮度 + 色温双指标分别统计后 merge(on="时段") 合并；多色柱状图（test_scripts/run_lighting_pipeline.py） | ✅ 通过 |
test_scripts/test_health.py | test_output_structure() 专项回归：AST 检查 run_pipeline.py 无硬编码盘符；test_scripts/ 目录无任何产物文件 | ✅ 通过 |
