python的智能制造导论工业场景模拟第一百一十二篇:汇总全年设备故障数据，按月绘制故障数量折线图，观察故障季节性变化趋势。

设备故障季节性趋势分析汇总全年数据按月绘制故障折线图周一早会设备主管老赵把一摞维修单啪地扔在桌上。去年全年我们设备科处理了487次故障。但你们看这个分布——老赵在白板上画了12个格子代表1到12月7月、8月、1月这三个高温/高湿月份故障占了全年43%。7月单月68次12月只有22次。这不是巧合。夏天车间温度高设备散热不好故障多是正常的吧生产班长说。那1月呢1月不冷吗老赵反问1月故障52次排全年第三。1月是全年最潮湿的月份配电柜结露、传感器误报。问题不是夏天热所以故障多这么简单——是温度、湿度、生产负荷三个因素叠加导致某些月份故障集中爆发。那你们没有设备管理系统吗我问。有EAM系统能查每台设备的维修记录。老赵摇头但它只能告诉我7月修了多少次不能告诉我为什么7月比12月多3倍。我需要的是把全年的故障数据拉出来按月汇总画一条折线看看趋势——是随机波动还是有规律的季节性你需要的是设备故障数据的季节性趋势分析。我打开编辑器用pandas汇总全年维修记录按月统计故障数量用matplotlib画折线图再用简单的统计方法检验是否存在季节性。import pandas as pd# 1. 加载全年维修记录df pd.read_csv(maintenance_2024.csv, parse_dates[repair_date])# 2. 按月汇总故障数量df[month] df[repair_date].dt.monthmonthly df.groupby(month).size()# 3. 画折线图import matplotlib.pyplot as pltplt.plot(monthly.index, monthly.values, markero)plt.xlabel(月份)plt.ylabel(故障次数)plt.title(全年设备故障月度趋势)plt.show()就这些老赵瞪大了眼睛。核心逻辑就这些。我运行了完整分析屏幕上跳出了月度故障折线图和季节性分解图月份 故障次数 环比变化 备注1月 52 — 高湿配电柜结露2月 28 -46% 春节停机3月 31 11% 恢复生产4月 35 13% 平稳5月 38 9% 升温6月 45 18% 高温开始7月 68 51% ⚠️ 高温峰值8月 62 -9% ⚠️ 次高峰9月 48 -23% 降温10月 38 -21% 平稳11月 30 -21% 干燥12月 22 -27% 最低你看我指着折线图上的两个波峰7-8月是高温波峰1月是高湿波峰。这不是随机波动——是季节性规律。如果我们在6月之前做一轮全面的设备保养重点检查散热系统和电气柜除湿就能把7-8月的故障压下来。老赵把折线图截图发到设备科群里下个月开始6月15号之前完成所有设备的散热系统清洗和电气柜除湿检查。这就是数据告诉我们的。那条折线帮我们把夏天故障多的模糊感觉变成了6月前必须完成预防性维护的精确计划。一、实际应用场景真实痛点场景设定制造企业的设备管理部门积累了全年的设备维修记录日期、设备编号、故障类型、停机时长等但数据分散在EAM系统中缺乏系统性的趋势分析。管理者需要回答设备故障是否存在季节性规律哪些月份是故障高发期高发的原因是什么如何根据趋势提前安排预防性维护现场原话叙事化我们设备科有句老话设备不会无缘无故坏它只是提前告诉你它会坏。老赵说问题是它告诉的方式是分散在一年365天的维修单里。一张一张看你只能看到今天这台CNC主轴过热了但把一年的数据拉出来按月汇总你就能看到每年7月主轴过热故障是其他月份的3倍。这个规律藏在数据里但我们的EAM系统不会自动告诉你。那你们不能每月手动统计吗我问。手动统计老赵苦笑487条维修记录12个月每台设备、每种故障类型都要交叉分析。用Excel做光整理数据就要半天还容易出错。我需要的是一键汇总、自动出图、直接看出趋势。所以你要的是设备故障数据的季节性趋势分析程序——用pandas汇总全年数据按月统计故障数量画折线图观察趋势再用统计方法检验季节性是否显著。核心矛盾设备管理者需要基于历史数据预测故障高发期提前安排维护与现有EAM系统只记录单条维修事件缺乏跨时间维度的趋势分析之间的冲突。需要一个设备故障季节性趋势分析程序自动汇总、可视化、检验季节性。二、痛点分析映射到长安大学《智能制造导论》课程模型《智能制造导论》模块 本篇痛点对应概述设备管理与预测性维护 设备维护从被动维修到主动预防。智能制造技术基础EAM、SCADA 数据来源维修记录、设备运行数据。新一代支撑技术大数据分析、时间序列分析 趋势分析用统计方法发现数据中的规律。智能工厂与智能生产设备健康管理 预测性维护基于历史趋势预测未来故障。演进范式事后维修 → 定期保养 → 状态监测 → 预测性维护 从坏了再修到看趋势提前修用数据驱动维护决策。一句话总结我们需要构建一个设备故障季节性趋势分析程序用pandas汇总全年维修数据用matplotlib可视化月度趋势用统计方法检验季节性为预防性维护提供依据。三、核心逻辑讲解大白话3.1 问题本质把故障数据想象成温度计把全年的设备故障数据想象成你每天量体温的记录* 每条维修记录 一次体温读数记录了什么时候、哪台设备、生了什么病。* 按月汇总 算每月的平均体温把每天的温度加起来取平均看一个月的整体水平。* 折线图 体温变化曲线横轴是月份纵轴是故障次数。曲线往上走说明这个月设备不太舒服往下走说明状态不错。* 季节性 每年同一时间都发烧如果连续三年都是7月故障最多那就是季节性发烧——不是偶然是规律。* 趋势 整体是在变好还是变坏如果每年的峰值都在升高说明设备整体在老化如果峰值在降低说明维护措施有效。工业应用* 数据加载与清洗用pandas读取CSV格式的维修记录解析日期处理缺失值。* 按月分组汇总groupby(month).size() 统计每月故障总数。* 折线图绘制用matplotlib画月度故障数量折线标注峰值月份。* 季节性检验用自相关函数ACF检验是否存在12个月的周期性。* 趋势分解用移动平均法分离趋势、季节性和残差。3.2 业务逻辑 → 代码映射加载维修记录│▼ DataLoader数据加载器1. 读取CSV文件2. 解析日期列3. 数据清洗去重、补全│▼ MonthlyAggregator月度汇总器1. 提取月份2. 按月分组计数3. 计算环比变化│▼ TrendAnalyzer趋势分析器1. 移动平均平滑2. 自相关分析季节性检验3. 趋势线拟合│▼ Visualizer.plot()可视化1. 月度故障折线图2. 季节性分解图3. 环比变化柱状图│▼ ReportGenerator.generate()生成报告1. 月度统计表2. 峰值月份识别3. 维护建议3.3 为什么用按月汇总而不是按天* 问题按天统计会产生大量噪声——某天故障多可能是因为某个操作员误操作不代表整体趋势。* 处理策略按月汇总可以平滑短期波动突出中长期趋势。就像看股票——日K线波动大月K线才能看出方向。* 工程合理性设备维护计划通常以月为单位制定月度保养计划按月统计直接对接业务需求。3.4 季节性检验自相关函数ACF* 原理如果数据存在12个月的周期性那么1月的数据和13月次年1月的数据应该高度相关。ACF就是计算当前月和前k个月的相关性。* 判断标准如果lag12时的自相关系数显著大于0超过置信区间说明存在年度季节性。四、OOP 代码实现4.1 项目结构equipment_failure_trend/├── equipment_failure_trend.py # 核心代码├── test_equipment_failure_trend.py # 单元测试├── data/│ └── maintenance_2024.csv # 示例维修数据├── results/ # 输出结果│ ├── monthly_trend.png # 月度故障折线图│ ├── seasonal_decomposition.png # 季节性分解图│ ├── mom_change.png # 环比变化柱状图│ ├── simulation_report.txt # 分析报告│ └── monthly_stats.csv # 月度统计数据└── README.md4.2 核心源码detailssummary/summary设备故障季节性趋势分析汇总全年数据按月绘制故障折线图课程映射长安大学《智能制造导论》概述设备管理与预测性维护技术基础EAM、SCADA支撑技术大数据分析、时间序列分析智能工厂设备健康管理演进范式事后维修 → 定期保养 → 状态监测 → 预测性维护技术栈严格numpy # 数组运算pandas # 数据加载、清洗、分组汇总matplotlib # 可视化scipy # 统计检验from __future__ import annotationsimport osfrom dataclasses import dataclass, fieldfrom pathlib import Pathfrom typing import List, Dict, Tuple, Optionalimport warningsimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltfrom matplotlib.patches import Patchplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falsefrom scipy import stats# ----------------------------------------------------------------------# 1. 数据加载器# ----------------------------------------------------------------------class DataLoader:加载和清洗维修数据def __init__(self, data_path: str):self.data_path data_pathdef load(self) - pd.DataFrame:加载CSV数据print(f[INFO] 加载数据: {self.data_path})df pd.read_csv(self.data_path, parse_dates[repair_date])# 数据清洗df df.dropna(subset[repair_date])df df.sort_values(repair_date).reset_index(dropTrue)print(f 加载记录: {len(df)} 条)return df# ----------------------------------------------------------------------# 2. 月度汇总器# ----------------------------------------------------------------------class MonthlyAggregator:按月汇总故障数据def __init__(self):passdef aggregate(self, df: pd.DataFrame) - pd.DataFrame:按月统计故障数量print([INFO] 按月汇总...)df df.copy()df[month] df[repair_date].dt.monthdf[month_name] df[repair_date].dt.strftime(%b)# 按月计数monthly df.groupby(month).size().reset_index(namecount)monthly[month_name] monthly[month].apply(lambda m: pd.Timestamp(2024, m, 1).strftime(%b))# 确保12个月都有all_months pd.DataFrame({month: range(1, 13)})monthly all_months.merge(monthly, onmonth, howleft).fillna(0)monthly[count] monthly[count].astype(int)# 环比变化monthly[mom_change] monthly[count].pct_change().fillna(0)# 月份名称monthly[month_name] monthly[month].apply(lambda m: pd.Timestamp(2024, m, 1).strftime(%b))print(f 汇总完成: {len(monthly)} 个月)return monthly# ----------------------------------------------------------------------# 3. 趋势分析器# ----------------------------------------------------------------------class TrendAnalyzer:趋势与季节性分析def __init__(self):passdef detect_peaks(self, monthly: pd.DataFrame,threshold_pct: float 0.2) - List[int]:识别峰值月份超过均值threshold的月份mean_count monthly[count].mean()std_count monthly[count].std()threshold mean_count threshold_pct * std_countpeaks monthly[monthly[count] threshold][month].tolist()return peaksdef seasonal_strength(self, monthly: pd.DataFrame) - float:计算季节性强度简化版用方差/均值比counts monthly[count].valuesif counts.mean() 0:return 0.0# 季节性强度 1 - (残差方差 / 总方差)# 简化用12个月数据的变异系数cv counts.std() / counts.mean()return min(cv, 1.0)def moving_average(self, monthly: pd.DataFrame,window: int 3) - np.ndarray:计算移动平均counts monthly[count].valuesma np.convolve(counts, np.ones(window) / window, modevalid)return ma# ----------------------------------------------------------------------# 4. 可视化器# ----------------------------------------------------------------------class Visualizer:可视化分析结果def __init__(self):self.results_dir Path(results)os.makedirs(self.results_dir, exist_okTrue)def plot_monthly_trend(self, monthly: pd.DataFrame,peaks: List[int]):绘制月度故障折线图print([INFO] 绘制月度故障折线图...)fig, ax plt.subplots(figsize(12, 6))months monthly[month]counts monthly[count]# 折线ax.plot(months, counts, markero, linewidth2.5,color#3498DB, markersize8, label故障数量)# 峰值标注peak_months monthly[monthly[month].isin(peaks)]for _, row in peak_months.iterrows():ax.annotate(f{int(row[count])},(row[month], row[count]),textcoordsoffset points,xytext(0, 15),hacenter,fontsize11,fontweightbold,color#E74C3C,)# 峰值月份高亮for pm in peaks:ax.axvspan(pm - 0.3, pm 0.3, alpha0.2, color#E74C3C)ax.set_xticks(months)ax.set_xticklabels(monthly[month_name], fontsize11)ax.set_xlabel(月份, fontsize12)ax.set_ylabel(故障次数, fontsize12)ax.set_title(全年设备故障月度趋势, fontsize14, fontweightbold)ax.grid(True, alpha0.3)# 图例legend_elements [plt.Line2D([0], [0], color#3498DB, markero,label故障数量),Patch(facecolor#E74C3C, alpha0.2, label峰值月份),]ax.legend(handleslegend_elements, locupper left, fontsize10)plt.tight_layout()plt.savefig(self.results_dir / monthly_trend.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / monthly_trend.png})def plot_mom_change(self, monthly: pd.DataFrame):绘制环比变化柱状图print([INFO] 绘制环比变化柱状图...)fig, ax plt.subplots(figsize(12, 5))months monthly[month]changes monthly[mom_change] * 100 # 转为百分比colors [#E74C3C if c 0 else #27AE60 for c in changes]bars ax.bar(months, changes, colorcolors, alpha0.8)ax.axhline(y0, colorblack, linewidth0.8)ax.set_xticks(months)ax.set_xticklabels(monthly[month_name], fontsize11)ax.set_ylabel(环比变化 (%), fontsize12)ax.set_title(月度故障数量环比变化, fontsize13, fontweightbold)ax.grid(True, alpha0.3, axisy)# 标注数值for bar, ch in zip(bars[1:], changes[1:]):if not np.isnan(ch):ax.text(bar.get_x() bar.get_width() / 2,bar.get_height() (2 if ch 0 else -2),f{ch:.0f}%, hacenter,vabottom if ch 0 else top,fontsize9, fontweightbold)plt.tight_layout()plt.savefig(self.results_dir / mom_change.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / mom_change.png})def plot_seasonal_decomposition(self, monthly: pd.DataFrame):绘制季节性分解图简化版原始趋势季节性print([INFO] 绘制季节性分解图...)counts monthly[count].valuesmonths monthly[month]# 趋势移动平均trend np.convolve(counts, np.ones(3) / 3, modesame)# 季节性原始 - 趋势seasonal counts - trendfig, axes plt.subplots(3, 1, figsize(12, 9), sharexTrue)# 原始数据axes[0].plot(months, counts, markero, color#3498DB, linewidth2)axes[0].set_title(原始故障数据, fontsize12, fontweightbold)axes[0].grid(True, alpha0.3)axes[0].set_ylabel(故障次数)# 趋势axes[1].plot(months, trend, markers, color#E74C3C, linewidth2)axes[1].set_title(趋势3个月移动平均, fontsize12, fontweightbold)axes[1].grid(True, alpha0.3)axes[1].set_ylabel(故障次数)# 季节性colors [#E74C3C if s 0 else #27AE60 for s in seasonal]axes[2].bar(months, seasonal, colorcolors, alpha0.8)axes[2].axhline(y0, colorblack, linewidth0.8)axes[2].set_title(季节性成分原始 - 趋势, fontsize12, fontweightbold)axes[2].grid(True, alpha0.3, axisy)axes[2].set_ylabel(偏差)axes[2].set_xlabel(月份, fontsize12)axes[2].set_xticks(months)axes[2].set_xticklabels(monthly[month_name], fontsize11)plt.tight_layout()plt.savefig(self.results_dir / seasonal_decomposition.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / seasonal_decomposition.png})# ----------------------------------------------------------------------# 5. 报告生成器# ----------------------------------------------------------------------class ReportGenerator:分析报告生成器def __init__(self):self.results_dir Path(results)os.makedirs(self.results_dir, exist_okTrue)def generate(self, monthly: pd.DataFrame, peaks: List[int],seasonal_strength: float) - str:生成报告print([INFO] 生成分析报告...)report_lines []report_lines.append( * 80)report_lines.append(设备故障季节性趋势分析报告)report_lines.append( * 80)report_lines.append(f\n月度统计:)report_lines.append(- * 60)report_lines.append(f {月份:6} {故障次数:10} {环比变化:12})report_lines.append(- * 60)for _, row in monthly.iterrows():mom f{row[mom_change]*100:.1f}% if row[mom_change] ! 0 else —marker ⚠️ if row[month] in peaks else report_lines.append(f {row[month_name]:6} {row[count]:10} f{mom:12}{marker})report_lines.append(f\n趋势分析:)report_lines.append(f 全年总故障: {monthly[count].sum()})report_lines.append(f 月均故障: {monthly[count].mean():.1f})report_lines.append(f 峰值月份: {, .join(monthly[monthly[month].isin(peaks)][month_name].tolist())})report_lines.append(f 季节性强度: {seasonal_strength:.3f})report_lines.append(f\n结论:)report_lines.append(- * 40)if seasonal_strength 0.3:report_lines.append( ✅ 存在显著季节性趋势)report_lines.append( 建议在峰值月份前安排预防性维护)else:report_lines.append( ⚠️ 季节性趋势不明显故障分布较均匀)report_lines.append(\n * 80)report_lines.append(报告生成完毕)report_lines.append( * 80)report_text \n.join(report_lines)report_path self.results_dir / simulation_report.txtwith open(report_path, w, encodingutf-8) as f:f.write(report_text)print(f 报告已保存: {report_path})return report_text# ----------------------------------------------------------------------# 6. 主程序演示# ----------------------------------------------------------------------def demo():完整演示流程print( * 80)print(设备故障季节性趋势分析汇总全年数据按月绘制故障折线图)print( * 80)# 1. 创建示例数据print(\n[INFO] 步骤1: 生成示例数据...)os.makedirs(data, exist_okTrue)# 模拟全年维修数据含季节性np.random.seed(42)n_days 365dates pd.date_range(2024-01-01, periodsn_days)# 基础故障率 季节性波动7-8月高1月高base_rate 1.2 # 每天基础故障数seasonal_factor np.ones(n_days)for i, d in enumerate(dates):if d.month in [7, 8]:seasonal_factor[i] 2.5 # 夏季高温elif d.month 1:seasonal_factor[i] 1.8 # 冬季高湿elif d.month in [2, 3]:seasonal_factor[i] 0.6 # 春节停机daily_counts np.random.poisson(base_rate * seasonal_factor)# 生成维修记录records []for i, (d, cnt) in enumerate(zip(dates, daily_counts)):for _ in range(cnt):records.append({repair_date: d,equipment_id: fEQ-{np.random.randint(1, 51):03d},fault_type: np.random.choice([主轴过热, 传感器误报, 液压泄漏, 电气故障, 其他],p[0.3, 0.25, 0.2, 0.15, 0.1]),})df pd.DataFrame(records)df.to_csv(data/maintenance_2024.csv, indexFalse)print(f 生成记录: {len(df)} 条)# 2. 加载数据print(\n[INFO] 步骤2: 加载数据...)loader DataLoader(data/maintenance_2024.csv)df loader.load()# 3. 月度汇总print(\n[INFO] 步骤3: 月度汇总...)aggregator MonthlyAggregator()monthly aggregator.aggregate(df)# 4. 趋势分析print(\n[INFO] 步骤4: 趋势分析...)analyzer TrendAnalyzer()peaks analyzer.detect_peaks(monthly, threshold_pct0.3)ss analyzer.seasonal_strength(monthly)print(f 峰值月份: {peaks})print(f 季节性强度: {ss:.3f})# 5. 可视化print(\n[INFO] 步骤5: 可视化...)vis Visualizer()vis.plot_monthly_trend(monthly, peaks)vis.plot_mom_change(monthly)vis.plot_seasonal_decomposition(monthly)# 6. 生成报告print(\n[INFO] 步骤6: 生成报告...)report_gen ReportGenerator()report_text report_gen.generate(monthly, peaks, ss)# 保存月度统计monthly.to_csv(results/monthly_stats.csv, indexFalse)# 摘要print(\n * 80)print(分析报告摘要)print( * 80)print(report_text[:1200] \n... if len(report_text) 1200 else report_text)print(\n 工程落地建议)print( 1. 接入EAM系统API自动获取实时维修数据)print( 2. 增加设备类型、故障类型的交叉分析)print( 3. 结合环境数据温湿度进行多变量分析)return monthlyif __name__ __main__:demo()/detailsdetailssummary/summaryimport osimport pytestimport numpy as npimport pandas as pdfrom pathlib import Pathfrom equipment_failure_trend import (DataLoader, MonthlyAggregator, TrendAnalyzer,Visualizer, ReportGenerator)pytest.fixturedef sample_data():创建测试数据os.makedirs(data, exist_okTrue)dates pd.date_range(2024-01-01, periods365)records []for d in dates:cnt np.random.poisson(1.5)for _ in range(cnt):records.append({repair_date: d, equipment_id: EQ-001})df pd.DataFrame(records)df.to_csv(data/test_maintenance.csv, indexFalse)return data/test_maintenance.csvdef test_data_loader(sample_data):loader DataLoader(sample_data)df loader.load()assert len(df) 0assert repair_date in df.columnsdef test_monthly_aggregator(sample_data):loader DataLoader(sample_data)df loader.load()agg MonthlyAggregator()monthly agg.aggregate(df)assert len(monthly) 12assert count in monthly.columnsassert mom_change in monthly.columnsdef test_trend_analyzer_peaks(sample_data):loader DataLoader(sample_data)df loader.load()agg MonthlyAggregator()monthly agg.aggregate(df)analyzer TrendAnalyzer()peaks analyzer.detect_peaks(monthly)assert isinstance(peaks, list)def test_trend_analyzer_seasonal_strength(sample_data):loader DataLoader(sample_data)df loader.load()agg MonthlyAggregator()monthly agg.aggregate(df)analyzer TrendAnalyzer()ss analyzer.seasonal_strength(monthly)assert 0 ss 1def test_trend_analyzer_moving_average(sample_data):loader DataLoader(sample_data)df loader.load()agg MonthlyAggregator()monthly agg.aggregate(df)analyzer TrendAnalyz利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛