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 即高风险比例。

常见错误与处理方式

常用分层边界（已验证）

• 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 顺手给出分母分子

已验证测试案例清单