企业官网建设流程全解析

设备故障季节性趋势分析：汇总全年数据，按月绘制故障折线图

周一早会，设备主管老赵把一摞维修单"啪"地扔在桌上。

"去年全年，我们设备科处理了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.month

monthly = df.groupby("month").size()

# 3. 画折线图

import matplotlib.pyplot as plt

plt.plot(monthly.index, monthly.values, marker="o")

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个月的相关性"。

* 判断标准：如果lag=12时的自相关系数显著大于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.md

4.2 核心源码

<details>

<summary></summary>

"""

设备故障季节性趋势分析：汇总全年数据，按月绘制故障折线图

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

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

概述：设备管理与预测性维护

技术基础：EAM、SCADA

支撑技术：大数据分析、时间序列分析

智能工厂：设备健康管理

演进范式：事后维修 → 定期保养 → 状态监测 → 预测性维护

技术栈（严格）：

numpy # 数组运算

pandas # 数据加载、清洗、分组汇总

matplotlib # 可视化

scipy # 统计检验

"""

from __future__ import annotations

import os

from dataclasses import dataclass, field

from pathlib import Path

from typing import List, Dict, Tuple, Optional

import warnings

import numpy as np

import pandas as pd

import matplotlib.pyplot as plt

from matplotlib.patches import Patch

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

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

from scipy import stats

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

# 1. 数据加载器

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

class DataLoader:

"""加载和清洗维修数据"""

def __init__(self, data_path: str):

self.data_path = data_path

def 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(drop=True)

print(f" 加载记录: {len(df)} 条")

return df

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

# 2. 月度汇总器

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

class MonthlyAggregator:

"""按月汇总故障数据"""

def __init__(self):

pass

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

"""按月统计故障数量"""

print("[INFO] 按月汇总...")

df = df.copy()

df["month"] = df["repair_date"].dt.month

df["month_name"] = df["repair_date"].dt.strftime("%b")

# 按月计数

monthly = df.groupby("month").size().reset_index(name="count")

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, on="month", how="left").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):

pass

def 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_count

peaks = monthly[monthly["count"] > threshold]["month"].tolist()

return peaks

def seasonal_strength(self, monthly: pd.DataFrame) -> float:

"""计算季节性强度（简化版：用方差/均值比）"""

counts = monthly["count"].values

if 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"].values

ma = np.convolve(counts, np.ones(window) / window, mode="valid")

return ma

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

# 4. 可视化器

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

class Visualizer:

"""可视化分析结果"""

def __init__(self):

self.results_dir = Path("results")

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

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, marker="o", linewidth=2.5,

color="#3498DB", markersize=8, label="故障数量")

# 峰值标注

peak_months = monthly[monthly["month"].isin(peaks)]

for _, row in peak_months.iterrows():

ax.annotate(

f"{int(row['count'])}",

(row["month"], row["count"]),

textcoords="offset points",

xytext=(0, 15),

ha="center",

fontsize=11,

fontweight="bold",

color="#E74C3C",

)

# 峰值月份高亮

for pm in peaks:

ax.axvspan(pm - 0.3, pm + 0.3, alpha=0.2, color="#E74C3C")

ax.set_xticks(months)

ax.set_xticklabels(monthly["month_name"], fontsize=11)

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

ax.set_ylabel("故障次数", fontsize=12)

ax.set_title("全年设备故障月度趋势", fontsize=14, fontweight="bold")

ax.grid(True, alpha=0.3)

# 图例

legend_elements = [

plt.Line2D([0], [0], color="#3498DB", marker="o",

label="故障数量"),

Patch(facecolor="#E74C3C", alpha=0.2, label="峰值月份"),

]

ax.legend(handles=legend_elements, loc="upper left", fontsize=10)

plt.tight_layout()

plt.savefig(self.results_dir / "monthly_trend.png",

dpi=150, bbox_inches="tight")

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, color=colors, alpha=0.8)

ax.axhline(y=0, color="black", linewidth=0.8)

ax.set_xticks(months)

ax.set_xticklabels(monthly["month_name"], fontsize=11)

ax.set_ylabel("环比变化 (%)", fontsize=12)

ax.set_title("月度故障数量环比变化", fontsize=13, fontweight="bold")

ax.grid(True, alpha=0.3, axis="y")

# 标注数值

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}%", ha="center",

va="bottom" if ch > 0 else "top",

fontsize=9, fontweight="bold")

plt.tight_layout()

plt.savefig(self.results_dir / "mom_change.png",

dpi=150, bbox_inches="tight")

plt.close()

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

def plot_seasonal_decomposition(self, monthly: pd.DataFrame):

"""绘制季节性分解图（简化版：原始+趋势+季节性）"""

print("[INFO] 绘制季节性分解图...")

counts = monthly["count"].values

months = monthly["month"]

# 趋势（移动平均）

trend = np.convolve(counts, np.ones(3) / 3, mode="same")

# 季节性（原始 - 趋势）

seasonal = counts - trend

fig, axes = plt.subplots(3, 1, figsize=(12, 9), sharex=True)

# 原始数据

axes[0].plot(months, counts, marker="o", color="#3498DB", linewidth=2)

axes[0].set_title("原始故障数据", fontsize=12, fontweight="bold")

axes[0].grid(True, alpha=0.3)

axes[0].set_ylabel("故障次数")

# 趋势

axes[1].plot(months, trend, marker="s", color="#E74C3C", linewidth=2)

axes[1].set_title("趋势（3个月移动平均）", fontsize=12, fontweight="bold")

axes[1].grid(True, alpha=0.3)

axes[1].set_ylabel("故障次数")

# 季节性

colors = ["#E74C3C" if s > 0 else "#27AE60" for s in seasonal]

axes[2].bar(months, seasonal, color=colors, alpha=0.8)

axes[2].axhline(y=0, color="black", linewidth=0.8)

axes[2].set_title("季节性成分（原始 - 趋势）", fontsize=12, fontweight="bold")

axes[2].grid(True, alpha=0.3, axis="y")

axes[2].set_ylabel("偏差")

axes[2].set_xlabel("月份", fontsize=12)

axes[2].set_xticks(months)

axes[2].set_xticklabels(monthly["month_name"], fontsize=11)

plt.tight_layout()

plt.savefig(self.results_dir / "seasonal_decomposition.png",

dpi=150, bbox_inches="tight")

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_ok=True)

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.txt"

with open(report_path, "w", encoding="utf-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_ok=True)

# 模拟全年维修数据（含季节性）

np.random.seed(42)

n_days = 365

dates = pd.date_range("2024-01-01", periods=n_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": f"EQ-{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", index=False)

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_pct=0.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", index=False)

# 摘要

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 monthly

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 equipment_failure_trend import (

DataLoader, MonthlyAggregator, TrendAnalyzer,

Visualizer, ReportGenerator

)

@pytest.fixture

def sample_data():

"""创建测试数据"""

os.makedirs("data", exist_ok=True)

dates = pd.date_range("2024-01-01", periods=365)

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", index=False)

return "data/test_maintenance.csv"

def test_data_loader(sample_data):

loader = DataLoader(sample_data)

df = loader.load()

assert len(df) > 0

assert "repair_date" in df.columns

def test_monthly_aggregator(sample_data):

loader = DataLoader(sample_data)

df = loader.load()

agg = MonthlyAggregator()

monthly = agg.aggregate(df)

assert len(monthly) == 12

assert "count" in monthly.columns

assert "mom_change" in monthly.columns

def 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 <= 1

def test_trend_analyzer_moving_average(sample_data):

loader = DataLoader(sample_data)

df = loader.load()

agg = MonthlyAggregator()

monthly = agg.aggregate(df)

analyzer = TrendAnalyz

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