data-pipeline/
├── pipeline/
│   ├── __init__.py
│   ├── collector.py    # 数据采集
│   ├── cleaner.py      # 数据清洗
│   ├── analyzer.py     # 数据分析
│   └── visualizer.py   # 数据可视化
├── config.py           # 配置
├── main.py             # 主流程
└── output/             # 输出目录

# pipeline/collector.py
import pandas as pd
import requests
from pathlib import Path

def collect_from_csv(file_path: str) -> pd.DataFrame:
    """从 CSV 文件采集数据"""
    return pd.read_csv(file_path)

def collect_from_api(url: str, params: dict = None) -> pd.DataFrame:
    """从 API 采集数据"""
    response = requests.get(url, params=params)
    response.raise_for_status()
    return pd.DataFrame(response.json())

def collect_from_database(connection_string: str, query: str) -> pd.DataFrame:
    """从数据库采集数据"""
    from sqlalchemy import create_engine
    engine = create_engine(connection_string)
    return pd.read_sql(query, engine)

# pipeline/cleaner.py
import pandas as pd
import numpy as np

def clean_sales_data(df: pd.DataFrame) -> pd.DataFrame:
    """清洗销售数据"""
    # 复制数据，避免修改原数据
    df = df.copy()

    # 1. 删除重复行
    df = df.drop_duplicates()

    # 2. 处理缺失值
    df["quantity"] = df["quantity"].fillna(0)
    df["price"] = df["price"].fillna(df["price"].median())
    df["customer_name"] = df["customer_name"].fillna("未知客户")

    # 3. 数据类型转换
    df["date"] = pd.to_datetime(df["date"])
    df["quantity"] = df["quantity"].astype(int)
    df["price"] = df["price"].astype(float)

    # 4. 计算衍生字段
    df["total"] = df["quantity"] * df["price"]
    df["month"] = df["date"].dt.to_period("M")
    df["weekday"] = df["date"].dt.day_name()

    # 5. 过滤异常值
    df = df[df["price"] > 0]
    df = df[df["quantity"] >= 0]

    return df

# pipeline/analyzer.py
import pandas as pd

def analyze_monthly_sales(df: pd.DataFrame) -> pd.DataFrame:
    """分析月度销售趋势"""
    monthly = df.groupby("month").agg(
        total_sales=("total", "sum"),
        order_count=("total", "count"),
        avg_order_value=("total", "mean")
    ).reset_index()

    monthly["growth_rate"] = monthly["total_sales"].pct_change() * 100
    return monthly

def analyze_top_products(df: pd.DataFrame, top_n: int = 10) -> pd.DataFrame:
    """分析热销产品"""
    return df.groupby("product").agg(
        total_quantity=("quantity", "sum"),
        total_revenue=("total", "sum")
    ).sort_values("total_revenue", ascending=False).head(top_n).reset_index()

def analyze_customer_segments(df: pd.DataFrame) -> pd.DataFrame:
    """客户分群分析"""
    customer_stats = df.groupby("customer_name").agg(
        total_spent=("total", "sum"),
        order_count=("total", "count"),
        avg_order_value=("total", "mean")
    ).reset_index()

    # 使用分位数进行分群
    customer_stats["segment"] = pd.qcut(
        customer_stats["total_spent"],
        q=3,
        labels=["低价值", "中价值", "高价值"]
    )

    return customer_stats

# pipeline/visualizer.py
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use("Agg")  # 非交互模式
plt.rcParams["font.sans-serif"] = ["SimHei"]  # 中文支持
plt.rcParams["axes.unicode_minus"] = False

def plot_monthly_trend(monthly_data, output_path: str):
    """绘制月度销售趋势图"""
    fig, ax1 = plt.subplots(figsize=(12, 6))

    # 柱状图：销售额
    ax1.bar(range(len(monthly_data)), monthly_data["total_sales"],
            color="steelblue", alpha=0.7, label="销售额")
    ax1.set_xlabel("月份")
    ax1.set_ylabel("销售额", color="steelblue")
    ax1.tick_params(axis="y", labelcolor="steelblue")

    # 折线图：增长率
    ax2 = ax1.twinx()
    ax2.plot(range(len(monthly_data)), monthly_data["growth_rate"],
             color="red", marker="o", label="增长率")
    ax2.set_ylabel("增长率 (%)", color="red")
    ax2.tick_params(axis="y", labelcolor="red")

    plt.title("月度销售趋势")
    fig.tight_layout()
    plt.savefig(output_path, dpi=150, bbox_inches="tight")
    plt.close()

def plot_top_products(products_data, output_path: str):
    """绘制热销产品图"""
    fig, ax = plt.subplots(figsize=(10, 6))

    ax.barh(products_data["product"], products_data["total_revenue"],
            color="teal")
    ax.set_xlabel("总收入")
    ax.set_title("Top 10 热销产品")

    plt.tight_layout()
    plt.savefig(output_path, dpi=150, bbox_inches="tight")
    plt.close()

# main.py
from pipeline.collector import collect_from_csv
from pipeline.cleaner import clean_sales_data
from pipeline.analyzer import analyze_monthly_sales, analyze_top_products
from pipeline.visualizer import plot_monthly_trend, plot_top_products
from pathlib import Path

def run_pipeline(input_file: str, output_dir: str = "output"):
    """运行完整的数据分析 Pipeline"""
    # 创建输出目录
    Path(output_dir).mkdir(exist_ok=True)

    print("📊 开始数据分析 Pipeline...")

    # 1. 数据采集
    print("  1. 采集数据...")
    raw_data = collect_from_csv(input_file)
    print(f"     采集到 {len(raw_data)} 条记录")

    # 2. 数据清洗
    print("  2. 清洗数据...")
    clean_data = clean_sales_data(raw_data)
    print(f"     清洗后 {len(clean_data)} 条记录")

    # 3. 数据分析
    print("  3. 分析数据...")
    monthly = analyze_monthly_sales(clean_data)
    top_products = analyze_top_products(clean_data)

    # 4. 数据可视化
    print("  4. 生成可视化...")
    plot_monthly_trend(monthly, f"{output_dir}/monthly_trend.png")
    plot_top_products(top_products, f"{output_dir}/top_products.png")

    # 5. 导出报告
    print("  5. 导出报告...")
    monthly.to_csv(f"{output_dir}/monthly_report.csv", index=False)
    top_products.to_csv(f"{output_dir}/top_products.csv", index=False)

    print("✅ Pipeline 完成！")
    print(f"   输出目录: {output_dir}/")

if __name__ == "__main__":
    run_pipeline("data/sales.csv")

data-pipeline/
├── pipeline/
│   ├── __init__.py
│   ├── collector.py    # 数据采集（CSV/API/数据库）
│   ├── cleaner.py      # 数据清洗（去重/填充/转换）
│   ├── analyzer.py     # 数据分析（统计/分群/趋势）
│   └── visualizer.py   # 数据可视化（图表生成）
├── config.py           # 配置管理
├── main.py             # 主流程编排
└── output/             # 输出目录
    ├── monthly_trend.png
    ├── top_products.png
    ├── monthly_report.csv
    └── top_products.csv