用Python透视订单数据按产品类别统计月度需求趋势月初的产销协调会上计划部小张被生产总监点名了。“这个月客户订单又变了。”总监指着投影上的Excel表格“小张你告诉我哪类产品需求涨了哪类跌了我好安排产线排程。”小张翻了翻手里的报表“总监我这边只有所有订单的明细一共三千多行。按产品类别……我还没来得及汇总。”“没来得及”总监皱眉“那你现在能告诉我电机类产品上个月下了多少单比前个月多了还是少了”小张在键盘上敲了半天筛选、求和额头上渗出汗珠“电机类……上个月大概120万前个月是……98万多了大概20%。”“大概大概是多少”总监追问。“等我再算一下……”小张手忙脚乱地切换筛选条件。会议室里其他人开始窃窃私语。五分钟过去了小张还在和Excel公式搏斗。“行了。”总监摆摆手“下次开会我要看到按类别汇总的月度趋势不是三千行明细。你今天下班前能弄出来吗”小张苦笑“我试试……”会后小张找到我“哥帮我写个程序吧。每次手动在Excel里筛选、求和、做透视表太容易出错了。而且总监要看趋势我得做几个月的数据手动搞根本来不及。”“这其实就是个数据透视分组聚合的问题。”我打开电脑“订单数据按产品类别分组再按月份聚合统计需求数量或金额最后画成趋势图。用pandas的pivot_table几行就搞定了。”小张凑过来看import pandas as pdimport 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,indexproduct_category,columnsmonth,valuesamount,aggfuncsum,fill_value0)# 绘制趋势pivot.T.plot(kindline, 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. 趋势图保存为PNG2. 数据摘要增长最快/下滑最多的类别3. 排程建议3.3 为什么用透视表而不是多次groupby* 透视表pivot_table一步到位生成二维交叉表行是类别列是月份值是汇总结果。适合展示行×列的矩阵数据。* 多次groupby先按类别groupby得到每个类别的总需求再按月份groupby得到每个月的总需求——但无法直接得到每个类别每个月的交叉结果。* 工程选择透视表更直观输出结果直接可以画折线图每列是一条线。如果只需要一维分组如每个月的总需求用groupby就够了。3.4 如何处理新产品类别和月份缺失* 新产品类别数据中出现新的产品类别程序自动将其作为新行加入透视表无需修改代码。* 月份缺失某个月某类产品没有订单透视表中会出现NaN空值。用fill_value0将空值填充为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.txt4.2 核心源码detailssummary/summary订单需求趋势分析按产品类别统计月度需求变化课程映射长安大学《智能制造导论》概述生产计划与控制、按订单生产MTO技术基础ERP系统、数据采集订单数据支撑技术工业大数据订单数据分析、商业智能智能工厂柔性生产、需求预测演进范式手工台账 → ERP → 固定报表 → 自助式数据分析技术栈严格numpy # 数值计算pandas # 数据加载、透视表matplotlib # 可视化networkx # 无scikit-learn # 无预留需求预测scipy # 无torch # 无from __future__ import annotationsimport osfrom dataclasses import dataclassfrom pathlib import Pathfrom typing import List, Dict, Tuple, Optionalfrom datetime import datetime, timedeltaimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] False# ----------------------------------------------------------------------# 1. 配置# ----------------------------------------------------------------------dataclassclass TrendConfig:趋势分析配置data_dir: str dataresults_dir: str results# 数据文件data_file: str orders.csv# 列名date_col: str order_datecategory_col: str product_categoryamount_col: str amountquantity_col: str quantity# 分析参数n_months: int 12 # 分析最近N个月random_seed: int 42# ----------------------------------------------------------------------# 2. 数据加载器# ----------------------------------------------------------------------class DataLoader:订单数据加载器def __init__(self, config: TrendConfig):self.config configself.data_dir Path(config.data_dir)os.makedirs(self.data_dir, exist_okTrue)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(days365)dates [start_date timedelta(daysnp.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) % 12if cat 电机:trend 1 month_idx * 0.05 # 逐月增长5%elif cat 控制器:trend 1 0.3 * np.sin(month_idx / 2) # 季节性波动else:trend 1amount 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: fORD-{i1: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, indexFalse)print(f 生成数据: {len(df)}条)return dfdef load_data(self) - pd.DataFrame:加载订单数据print(f[INFO] 加载订单数据...)csv_path self.data_dir / self.config.data_fileif 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 configdef 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 configdef analyze(self, df: pd.DataFrame) - pd.DataFrame:分析月度需求趋势print(f[INFO] 分析月度需求趋势...)# 透视表类别 × 月份汇总金额pivot pd.pivot_table(df,indexself.config.category_col,columnsmonth,valuesself.config.amount_col,aggfuncsum,fill_value0)# 计算环比增长率如果有至少两个月数据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] * 100print(f 透视表形状: {pivot.shape})print(f 月份数: {len(pivot.columns)})return pivot# ----------------------------------------------------------------------# 5. 可视化器# ----------------------------------------------------------------------class Visualizer:可视化结果def __init__(self, config: TrendConfig):self.config configself.results_dir Path(config.results_dir)os.makedirs(self.results_dir, exist_okTrue)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], markero, linewidth2, labelidx)ax.set_xlabel(月份, fontsize12)ax.set_ylabel(订单金额元, fontsize12)ax.set_title(各类产品月度订单需求趋势, fontsize14, fontweightbold)ax.legend(fontsize11)ax.grid(True, alpha0.3)plt.xticks(rotation45)plt.tight_layout()plt.savefig(self.results_dir / demand_trend_line.png, dpi150, bbox_inchestight)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, labelspivot.index, alpha0.8)ax.set_xlabel(月份, fontsize12)ax.set_ylabel(订单金额元, fontsize12)ax.set_title(各类产品月度订单需求占比堆叠面积图, fontsize14, fontweightbold)ax.legend(fontsize11, locupper left)plt.xticks(rotation45)plt.tight_layout()plt.savefig(self.results_dir / demand_trend_area.png, dpi150, bbox_inchestight)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, cmapYlOrRd, aspectauto)ax.set_xticks(range(len(pivot.columns)))ax.set_xticklabels(pivot.columns, rotation45)ax.set_yticks(range(len(pivot.index)))ax.set_yticklabels(pivot.index)ax.set_title(需求热力图颜色越深需求越大, fontsize14, fontweightbold)plt.colorbar(im, axax, label订单金额元)plt.tight_layout()plt.savefig(self.results_dir / demand_heatmap.png, dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / demand_heatmap.png})# ----------------------------------------------------------------------# 6. 报告生成器# ----------------------------------------------------------------------class ReportGenerator:分析报告生成器def __init__(self, config: TrendConfig):self.config configself.results_dir Path(config.results_dir)os.makedirs(self.results_dir, exist_okTrue)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] * 100fastest 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.txtwith open(report_path, w, encodingutf-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_dirdata,results_dirresults)# 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 pivotif __name__ __main__:demo()/detailsdetailssummary/summaryimport osimport pytestimport numpy as npimport pandas as pdfrom pathlib import Pathfrom order_demand_trend import (TrendConfig, DataLoader, DataPreprocessor,TrendAnalyzer, Visualizer, ReportGenerator)pytest.fixturedef config():return TrendConfig(data_dirtest_data,results_dirtest_results)pytest.fixturedef loader(config):l DataLoader(config)yield lfor 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 12def test_data_loader_generate(config, loader):df loader.generate_synthetic_data(n_records500)assert len(df) 500def test_data_loader_load(config, loader):loader.generate_synthetic_data(n_records200)df loader.load_data()assert len(df) 0def test_preprocessor(config, loader):loader.generate_synthetic_data(n_records200)df loader.load_data()pre DataPreprocessor(config)processed pre.preprocess(df)assert month in processed.columnsdef test_analyzer(config, loader):loader.generate_synthetic_data(n_records500)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) 0def test_visualizer_line(config, loader):loader.generate_synthetic_data(n_records300)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_records300)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_records300)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_records300)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 reportdef test_end_to_end(config, loader):loader.generate_synthetic_data(n_records500)df loader.load_data()pre DataPreprocessor(config)processed pre.preproce利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
