企业官网建设流程全解析

用Python透视订单数据：按产品类别统计月度需求趋势

月初的产销协调会上，计划部小张被生产总监点名了。

“这个月客户订单又变了。”总监指着投影上的Excel表格，“小张，你告诉我，哪类产品需求涨了？哪类跌了？我好安排产线排程。”

小张翻了翻手里的报表：“总监，我这边只有所有订单的明细，一共三千多行。按产品类别……我还没来得及汇总。”

“没来得及？”总监皱眉，“那你现在能告诉我，电机类产品上个月下了多少单？比前个月多了还是少了？”

小张在键盘上敲了半天，筛选、求和，额头上渗出汗珠：“电机类……上个月大概120万，前个月是……98万？多了大概20%。”

“大概？大概是多少？”总监追问。

“等我再算一下……”小张手忙脚乱地切换筛选条件。

会议室里其他人开始窃窃私语。五分钟过去了，小张还在和Excel公式搏斗。

“行了。”总监摆摆手，“下次开会我要看到按类别汇总的月度趋势，不是三千行明细。你今天下班前能弄出来吗？”

小张苦笑：“我试试……”

会后，小张找到我：“哥，帮我写个程序吧。每次手动在Excel里筛选、求和、做透视表，太容易出错了。而且总监要看趋势，我得做几个月的数据，手动搞根本来不及。”

“这其实就是个数据透视+分组聚合的问题。”我打开电脑，“订单数据按产品类别分组，再按月份聚合，统计需求数量或金额，最后画成趋势图。用pandas的

"pivot_table"几行就搞定了。”

小张凑过来看：

import pandas as pd

import matplotlib.pyplot as plt

# 读取订单数据

orders = pd.read_csv("orders.csv", parse_dates=["order_date"])

# 提取月份

orders["month"] = orders["order_date"].dt.to_period("M")

# 透视表：按类别×月份汇总金额

pivot = pd.pivot_table(

orders,

index="product_category",

columns="month",

values="amount",

aggfunc="sum",

fill_value=0

)

# 绘制趋势

pivot.T.plot(kind="line", figsize=(12, 6))

plt.ylabel("订单金额（万元）")

plt.title("各类产品月度订单需求趋势")

“就这些？”小张瞪大了眼睛。

“核心就这么多。”我运行代码，屏幕上立刻出现了四条折线——电机类、控制器类、传感器类、结构件类，每条线代表一个产品类别在各个月的需求变化。

“你看。”我指着图，“电机类从1月到5月稳步上升，5月突然跳到150万；控制器类3月有个峰值，之后回落；传感器类一直平稳在30万左右。这些信息，你用Excel得做多少个透视表才能看出来？”

小张立刻把图拿给总监看。总监扫了一眼：“电机类涨了这么多，下个月产线排程重点放在电机上。传感器类平稳，维持现有节奏就行。”

五秒钟，决策做完了。

“以前我做透视表要半小时，还怕选错区域。”小张说，“现在几秒钟出图，而且数据更新后重新跑一遍就行，不用再手动折腾了。”

一、实际应用场景（真实痛点）

场景设定：工厂接获大量客户订单，每条订单记录产品名称、产品类别、下单日期、数量、金额等。计划部门需要按产品类别统计每月的订单需求总量，分析需求变化趋势，以指导生产排程和物料采购。但订单数据量大（数千行），手工在Excel中筛选、汇总、制作透视表效率低下且容易出错，难以快速响应管理层的决策需求。

现场原话（叙事化）：

“我们计划部有句老话：'订单明细是原材料，趋势图才是成品。'”小张说，“我每天面对几千行订单数据，但总监不关心每一行，他关心的是'哪类产品这个月涨了、哪类跌了'。我要把原材料加工成成品，但Excel这个'加工工具'太慢了。”

“那你们系统里没有报表吗？”我问。

“ERP系统有标准报表，但只能按固定维度汇总。”小张摇头，“我想看'按产品类别的月度趋势'，系统不支持。要么导出明细自己处理，要么找IT部定制报表——IT部的排期是两个月后。”

“那就自己用Python处理。”我说。

“可我不会写代码啊。”小张说。

“我帮你写个程序，你只要把ERP导出的CSV文件放进去，运行一下就出图。”我说，“以后每次有新数据，替换文件重新运行就行。”

核心矛盾："管理层需要按产品类别的月度需求趋势"与"现有ERP报表不支持自定义维度、手工Excel处理效率低下"之间的冲突。需要一个"订单需求趋势分析程序"，自动从订单明细中按类别和月份分组聚合，输出趋势图表。

二、痛点分析（映射到长安大学《智能制造导论》课程模型）

《智能制造导论》模块 本篇痛点对应

概述：生产计划与控制、按订单生产（MTO） 生产计划：基于订单需求趋势安排生产排程和物料需求计划（MRP）。

智能制造技术基础：ERP系统、数据采集（订单数据） ERP系统：订单数据来源于ERP，但分析能力有限，需要外部工具增强。

新一代支撑技术：工业大数据（订单数据分析）、商业智能（BI） 商业智能：从订单明细中挖掘需求模式，支持数据驱动的决策。

智能工厂与智能生产：柔性生产、需求预测 需求预测：基于历史订单趋势，预测未来需求，指导产能规划。

演进范式：手工台账 → ERP系统 → 固定报表 → 自助式数据分析（Python脚本） 从"依赖IT部门定制报表"到"业务人员自助分析"，实现从"滞后响应"到"实时洞察"的跨越。

一句话总结：我们需要构建一个"订单需求趋势分析程序"，从订单明细中按产品类别和月份分组聚合，用可视化展示需求变化趋势，为生产计划和物料采购提供数据支撑。

三、核心逻辑讲解（大白话）

3.1 问题本质：把订单数据看成"超市小票"，统计"每个月每类商品卖了多少"

把订单分析和超市购物的关系，想象成"超市经理看销售报表"：

* 每张订单 = 一张超市小票：小票上写着你买了什么（产品名称）、属于什么品类（产品类别）、买了几个（数量）、花了多少钱（金额）。

* 产品类别 = 商品分区：超市里有生鲜区、零食区、日化区。你想知道"这个月哪个区卖得最好"，而不是关心每一包薯片的具体价格。

* 月份 = 时间段：你想比较"一月 vs 二月 vs 三月"的销售变化，看趋势是涨是跌。

* 分组聚合 = 按区按月的收银台：把每个月每个区的所有小票金额加起来，就是这个区那个月的销售额。

* 趋势图 = 销售走势图：把每个月的数据连成线，一眼看出哪个区在增长、哪个区在下滑。

工业应用：

* 输入：订单明细CSV，包含字段：订单日期、产品名称、产品类别、数量、金额等。

* 提取月份：从订单日期中提取年份和月份（如"2026-03"），作为时间维度。

* 分组聚合：按产品类别和月份分组，对数量或金额求和。

* 透视表：行是产品类别，列是月份，值是汇总金额——这就是一张二维的需求矩阵。

* 趋势图：每类产品一条折线，X轴是月份，Y轴是需求金额，直观展示变化趋势。

3.2 业务逻辑 → 代码映射

读取订单数据CSV

│

▼ DataLoader.load()

数据加载：

1. pd.read_csv() 读取订单

2. parse_dates 解析订单日期

│

▼ DataPreprocessor.preprocess()

数据预处理：

1. 提取月份：dt.to_period("M") 或 dt.strftime("%Y-%m")

2. 去除异常订单（如数量为负、金额为0）

│

▼ TrendAnalyzer.analyze()

趋势分析：

1. 分组聚合：groupby(["product_category", "month"])["amount"].sum()

2. 透视表：pivot_table()

3. 计算环比增长率（可选）

│

▼ Visualizer.plot()

可视化：

1. 折线图：每类产品一条线，展示月度趋势

2. 堆叠面积图：展示各类别需求占比变化

3. 热力图：行列为类别×月份，颜色深浅表示需求大小

│

▼ ReportGenerator.generate_report()

生成报告：

1. 趋势图保存为PNG

2. 数据摘要（增长最快/下滑最多的类别）

3. 排程建议

3.3 为什么用透视表而不是多次groupby？

* 透视表（pivot_table）：一步到位生成二维交叉表，行是类别，列是月份，值是汇总结果。适合展示"行×列"的矩阵数据。

* 多次groupby：先按类别groupby得到每个类别的总需求，再按月份groupby得到每个月的总需求——但无法直接得到"每个类别每个月"的交叉结果。

* 工程选择：透视表更直观，输出结果直接可以画折线图（每列是一条线）。如果只需要一维分组（如"每个月的总需求"），用groupby就够了。

3.4 如何处理"新产品类别"和"月份缺失"？

* 新产品类别：数据中出现新的产品类别，程序自动将其作为新行加入透视表，无需修改代码。

* 月份缺失：某个月某类产品没有订单，透视表中会出现NaN（空值）。用

"fill_value=0"将空值填充为0，确保趋势图连续。

* 工程处理：在预处理阶段，可以用

"pd.date_range"生成完整的月份序列，用

"reindex"确保所有月份都出现在X轴上，即使某个月完全没有订单。

四、OOP 代码实现

4.1 项目结构

order_demand_trend/

├── data/

│ └── orders.csv # 订单明细数据

├── results/ # 可视化结果

│ ├── demand_trend_line.png

│ ├── demand_trend_area.png

│ ├── demand_heatmap.png

│ └── trend_report.txt

├── order_demand_trend.py # 核心代码

├── test_order_demand_trend.py # 单元测试

├── README.md

└── requirements.txt

4.2 核心源码

<details>

<summary></summary>

"""

订单需求趋势分析：按产品类别统计月度需求变化

=================================================================

课程映射（长安大学《智能制造导论》）：

概述：生产计划与控制、按订单生产（MTO）

技术基础：ERP系统、数据采集（订单数据）

支撑技术：工业大数据（订单数据分析）、商业智能

智能工厂：柔性生产、需求预测

演进范式：手工台账 → ERP → 固定报表 → 自助式数据分析

技术栈（严格）：

numpy # 数值计算

pandas # 数据加载、透视表

matplotlib # 可视化

networkx # 无

scikit-learn # 无（预留需求预测）

scipy # 无

torch # 无

"""

from __future__ import annotations

import os

from dataclasses import dataclass

from pathlib import Path

from typing import List, Dict, Tuple, Optional

from datetime import datetime, timedelta

import numpy as np

import pandas as pd

import matplotlib.pyplot as plt

plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]

plt.rcParams["axes.unicode_minus"] = False

# ----------------------------------------------------------------------

# 1. 配置

# ----------------------------------------------------------------------

@dataclass

class TrendConfig:

"""趋势分析配置"""

data_dir: str = "data"

results_dir: str = "results"

# 数据文件

data_file: str = "orders.csv"

# 列名

date_col: str = "order_date"

category_col: str = "product_category"

amount_col: str = "amount"

quantity_col: str = "quantity"

# 分析参数

n_months: int = 12 # 分析最近N个月

random_seed: int = 42

# ----------------------------------------------------------------------

# 2. 数据加载器

# ----------------------------------------------------------------------

class DataLoader:

"""订单数据加载器"""

def __init__(self, config: TrendConfig):

self.config = config

self.data_dir = Path(config.data_dir)

os.makedirs(self.data_dir, exist_ok=True)

def generate_synthetic_data(self, n_records: int = 3000):

"""生成模拟订单数据"""

print(f"[INFO] 生成模拟订单数据（{n_records}条）...")

np.random.seed(self.config.random_seed)

# 产品类别

categories = ["电机", "控制器", "传感器", "结构件", "线缆"]

# 时间范围：最近12个月

end_date = datetime.now()

start_date = end_date - timedelta(days=365)

dates = [start_date + timedelta(days=np.random.randint(0, 365))

for _ in range(n_records)]

# 生成订单

data = []

for i in range(n_records):

cat = np.random.choice(categories)

# 不同类别基础金额不同

base_amount = {

"电机": 5000, "控制器": 3000, "传感器": 800,

"结构件": 1200, "线缆": 200

}[cat]

# 月度趋势：电机增长，控制器波动，传感器平稳

month_idx = (dates[i].month + (dates[i].year - start_date.year) * 12) % 12

if cat == "电机":

trend = 1 + month_idx * 0.05 # 逐月增长5%

elif cat == "控制器":

trend = 1 + 0.3 * np.sin(month_idx / 2) # 季节性波动

else:

trend = 1

amount = base_amount * trend * np.random.uniform(0.8, 1.2)

quantity = max(1, int(amount / base_amount * np.random.uniform(0.5, 1.5)))

data.append({

"order_id": f"ORD-{i+1:06d}",

"order_date": dates[i],

"product_category": cat,

"product_name": f"{cat}-产品-{np.random.randint(1, 20):03d}",

"quantity": quantity,

"amount": round(amount, 2)

})

df = pd.DataFrame(data)

df.to_csv(self.data_dir / self.config.data_file, index=False)

print(f" 生成数据: {len(df)}条")

return df

def load_data(self) -> pd.DataFrame:

"""加载订单数据"""

print(f"[INFO] 加载订单数据...")

csv_path = self.data_dir / self.config.data_file

if not csv_path.exists():

self.generate_synthetic_data()

df = pd.read_csv(csv_path, parse_dates=[self.config.date_col])

print(f" 加载数据: {len(df)}条")

print(f" 时间范围: {df[self.config.date_col].min()} 至 {df[self.config.date_col].max()}")

print(f" 产品类别: {df[self.config.category_col].unique().tolist()}")

return df

# ----------------------------------------------------------------------

# 3. 数据预处理器

# ----------------------------------------------------------------------

class DataPreprocessor:

"""数据预处理"""

def __init__(self, config: TrendConfig):

self.config = config

def preprocess(self, df: pd.DataFrame) -> pd.DataFrame:

"""预处理数据"""

print(f"[INFO] 预处理数据...")

df = df.copy()

# 提取月份

df["month"] = df[self.config.date_col].dt.to_period("M").astype(str)

# 过滤异常

before = len(df)

df = df[(df[self.config.amount_col] > 0) & (df[self.config.quantity_col] > 0)]

after = len(df)

if before != after:

print(f" 去除异常: {before - after}条")

print(f" 预处理完成: {len(df)}条")

return df

# ----------------------------------------------------------------------

# 4. 趋势分析器

# ----------------------------------------------------------------------

class TrendAnalyzer:

"""趋势分析"""

def __init__(self, config: TrendConfig):

self.config = config

def analyze(self, df: pd.DataFrame) -> pd.DataFrame:

"""分析月度需求趋势"""

print(f"[INFO] 分析月度需求趋势...")

# 透视表：类别 × 月份，汇总金额

pivot = pd.pivot_table(

df,

index=self.config.category_col,

columns="month",

values=self.config.amount_col,

aggfunc="sum",

fill_value=0

)

# 计算环比增长率（如果有至少两个月数据）

if len(pivot.columns) >= 2:

for i in range(1, len(pivot.columns)):

prev_col = pivot.columns[i-1]

curr_col = pivot.columns[i]

pivot[f"{curr_col}_MoM"] = (pivot[curr_col] - pivot[prev_col]) / pivot[prev_col] * 100

print(f" 透视表形状: {pivot.shape}")

print(f" 月份数: {len(pivot.columns)}")

return pivot

# ----------------------------------------------------------------------

# 5. 可视化器

# ----------------------------------------------------------------------

class Visualizer:

"""可视化结果"""

def __init__(self, config: TrendConfig):

self.config = config

self.results_dir = Path(config.results_dir)

os.makedirs(self.results_dir, exist_ok=True)

def plot_trend_line(self, pivot: pd.DataFrame):

"""折线图：月度需求趋势"""

print(f"[INFO] 绘制折线图...")

fig, ax = plt.subplots(figsize=(14, 6))

for idx in pivot.index:

ax.plot(pivot.columns, pivot.loc[idx], marker="o", linewidth=2, label=idx)

ax.set_xlabel("月份", fontsize=12)

ax.set_ylabel("订单金额（元）", fontsize=12)

ax.set_title("各类产品月度订单需求趋势", fontsize=14, fontweight="bold")

ax.legend(fontsize=11)

ax.grid(True, alpha=0.3)

plt.xticks(rotation=45)

plt.tight_layout()

plt.savefig(self.results_dir / "demand_trend_line.png", dpi=150, bbox_inches="tight")

plt.close()

print(f" 已保存: {self.results_dir / 'demand_trend_line.png'}")

def plot_trend_area(self, pivot: pd.DataFrame):

"""堆叠面积图：需求占比变化"""

print(f"[INFO] 绘制堆叠面积图...")

fig, ax = plt.subplots(figsize=(14, 6))

ax.stackplot(pivot.columns, pivot.values, labels=pivot.index, alpha=0.8)

ax.set_xlabel("月份", fontsize=12)

ax.set_ylabel("订单金额（元）", fontsize=12)

ax.set_title("各类产品月度订单需求占比（堆叠面积图）", fontsize=14, fontweight="bold")

ax.legend(fontsize=11, loc="upper left")

plt.xticks(rotation=45)

plt.tight_layout()

plt.savefig(self.results_dir / "demand_trend_area.png", dpi=150, bbox_inches="tight")

plt.close()

print(f" 已保存: {self.results_dir / 'demand_trend_area.png'}")

def plot_heatmap(self, pivot: pd.DataFrame):

"""热力图：需求强度"""

print(f"[INFO] 绘制热力图...")

fig, ax = plt.subplots(figsize=(12, 5))

im = ax.imshow(pivot.values, cmap="YlOrRd", aspect="auto")

ax.set_xticks(range(len(pivot.columns)))

ax.set_xticklabels(pivot.columns, rotation=45)

ax.set_yticks(range(len(pivot.index)))

ax.set_yticklabels(pivot.index)

ax.set_title("需求热力图（颜色越深需求越大）", fontsize=14, fontweight="bold")

plt.colorbar(im, ax=ax, label="订单金额（元）")

plt.tight_layout()

plt.savefig(self.results_dir / "demand_heatmap.png", dpi=150, bbox_inches="tight")

plt.close()

print(f" 已保存: {self.results_dir / 'demand_heatmap.png'}")

# ----------------------------------------------------------------------

# 6. 报告生成器

# ----------------------------------------------------------------------

class ReportGenerator:

"""分析报告生成器"""

def __init__(self, config: TrendConfig):

self.config = config

self.results_dir = Path(config.results_dir)

os.makedirs(self.results_dir, exist_ok=True)

def generate_report(self, pivot: pd.DataFrame):

"""生成报告"""

print(f"[INFO] 生成分析报告...")

report_lines = []

report_lines.append("=" * 80)

report_lines.append("订单需求趋势分析报告")

report_lines.append("=" * 80)

report_lines.append(f"\n分析时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

# 概况

report_lines.append(f"\n数据概况:")

report_lines.append(f" 产品类别: {list(pivot.index)}")

report_lines.append(f" 月份: {list(pivot.columns)}")

# 各类别总需求

report_lines.append(f"\n各类别总需求:")

for cat in pivot.index:

total = pivot.loc[cat].sum()

report_lines.append(f" {cat}: {total:,.0f}元")

# 增长最快的类别

if len(pivot.columns) >= 2:

first_month = pivot.columns[0]

last_month = pivot.columns[-1]

growth = (pivot[last_month] - pivot[first_month]) / pivot[first_month] * 100

fastest = growth.idxmax()

slowest = growth.idxmin()

report_lines.append(f"\n增长分析（{first_month} → {last_month}）:")

report_lines.append(f" 增长最快: {fastest} ({growth[fastest]:+.1f}%)")

report_lines.append(f" 下滑最多: {slowest} ({growth[slowest]:+.1f}%)")

# 排程建议

report_lines.append(f"\n排程建议:")

report_lines.append("-" * 40)

if len(pivot.columns) >= 2:

latest_month = pivot.columns[-1]

top_category = pivot[latest_month].idxmax()

report_lines.append(f" 1. {latest_month}需求最高的类别是{top_category}，建议优先排产")

report_lines.append(f" 2. 关注{fastest}的增长趋势，提前准备产能和物料")

report_lines.append("\n" + "=" * 80)

report_lines.append("报告生成完毕")

report_lines.append("=" * 80)

report_text = "\n".join(report_lines)

report_path = self.results_dir / "trend_report.txt"

with open(report_path, "w", encoding="utf-8") as f:

f.write(report_text)

print(f" 报告已保存: {report_path}")

return report_text

# ----------------------------------------------------------------------

# 7. 主程序演示

# ----------------------------------------------------------------------

def demo():

"""完整演示流程"""

print("=" * 80)

print("订单需求趋势分析：按产品类别统计月度需求变化")

print("=" * 80)

config = TrendConfig(

data_dir="data",

results_dir="results"

)

# 1. 加载

print("\n[INFO] 步骤1: 加载数据...")

loader = DataLoader(config)

df = loader.load_data()

# 2. 预处理

print("\n[INFO] 步骤2: 预处理...")

preprocessor = DataPreprocessor(config)

df = preprocessor.preprocess(df)

# 3. 分析

print("\n[INFO] 步骤3: 分析趋势...")

analyzer = TrendAnalyzer(config)

pivot = analyzer.analyze(df)

# 4. 可视化

print("\n[INFO] 步骤4: 可视化...")

visualizer = Visualizer(config)

visualizer.plot_trend_line(pivot)

visualizer.plot_trend_area(pivot)

visualizer.plot_heatmap(pivot)

# 5. 报告

print("\n[INFO] 步骤5: 生成报告...")

report_gen = ReportGenerator(config)

report_text = report_gen.generate_report(pivot)

# 摘要

print("\n" + "=" * 80)

print("分析报告摘要")

print("=" * 80)

print(report_text[:1500] + "\n..." if len(report_text) > 1500 else report_text)

print("\n🔧 工程落地建议：")

print(" 1. 将脚本集成到ERP导出流程，每月自动生成趋势图")

print(" 2. 增加需求预测模块（ARIMA/LSTM）")

print(" 3. 结合库存数据，生成物料需求计划（MRP）")

return pivot

if __name__ == "__main__":

demo()

</details>

<details>

<summary></summary>

import os

import pytest

import numpy as np

import pandas as pd

from pathlib import Path

from order_demand_trend import (

TrendConfig, DataLoader, DataPreprocessor,

TrendAnalyzer, Visualizer, ReportGenerator

)

@pytest.fixture

def config():

return TrendConfig(

data_dir="test_data",

results_dir="test_results"

)

@pytest.fixture

def loader(config):

l = DataLoader(config)

yield l

for d in [Path("test_data"), Path("test_results")]:

if d.exists():

for f in d.iterdir():

f.unlink()

d.rmdir()

def test_config(config):

assert config.n_months == 12

def test_data_loader_generate(config, loader):

df = loader.generate_synthetic_data(n_records=500)

assert len(df) == 500

def test_data_loader_load(config, loader):

loader.generate_synthetic_data(n_records=200)

df = loader.load_data()

assert len(df) > 0

def test_preprocessor(config, loader):

loader.generate_synthetic_data(n_records=200)

df = loader.load_data()

pre = DataPreprocessor(config)

processed = pre.preprocess(df)

assert "month" in processed.columns

def test_analyzer(config, loader):

loader.generate_synthetic_data(n_records=500)

df = loader.load_data()

pre = DataPreprocessor(config)

processed = pre.preprocess(df)

analyzer = TrendAnalyzer(config)

pivot = analyzer.analyze(processed)

assert isinstance(pivot, pd.DataFrame)

assert len(pivot) > 0

def test_visualizer_line(config, loader):

loader.generate_synthetic_data(n_records=300)

df = loader.load_data()

pre = DataPreprocessor(config)

processed = pre.preprocess(df)

analyzer = TrendAnalyzer(config)

pivot = analyzer.analyze(processed)

vis = Visualizer(config)

vis.plot_trend_line(pivot)

assert Path("test_results/demand_trend_line.png").exists()

def test_visualizer_area(config, loader):

loader.generate_synthetic_data(n_records=300)

df = loader.load_data()

pre = DataPreprocessor(config)

processed = pre.preprocess(df)

analyzer = TrendAnalyzer(config)

pivot = analyzer.analyze(processed)

vis = Visualizer(config)

vis.plot_trend_area(pivot)

assert Path("test_results/demand_trend_area.png").exists()

def test_visualizer_heatmap(config, loader):

loader.generate_synthetic_data(n_records=300)

df = loader.load_data()

pre = DataPreprocessor(config)

processed = pre.preprocess(df)

analyzer = TrendAnalyzer(config)

pivot = analyzer.analyze(processed)

vis = Visualizer(config)

vis.plot_heatmap(pivot)

assert Path("test_results/demand_heatmap.png").exists()

def test_report_generator(config, loader):

loader.generate_synthetic_data(n_records=300)

df = loader.load_data()

pre = DataPreprocessor(config)

processed = pre.preprocess(df)

analyzer = TrendAnalyzer(config)

pivot = analyzer.analyze(processed)

report_gen = ReportGenerator(config)

report = report_gen.generate_report(pivot)

assert isinstance(report, str)

assert "需求" in report

def test_end_to_end(config, loader):

loader.generate_synthetic_data(n_records=500)

df = loader.load_data()

pre = DataPreprocessor(config)

processed = pre.preproce

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！