1. 实际应用场景描述
某物业公司或能源管理公司需要监控住户或商户的水、电、燃气使用情况，防止因为管道泄漏、设备故障或人为浪费造成资源损耗和额外费用。
传统做法是人工每月查看账单，发现异常时往往已经浪费了一段时间。
本工具的目标是自动化分析近 12 个月的缴费数据，计算月均用量，并在当月用量超过均值 20% 时自动发出预警，帮助管理人员及时排查隐患。
2. 痛点分析
- 人工检查效率低：大量用户数据无法逐一快速判断。
- 异常发现滞后：等到账单出来才发现，浪费已发生。
- 缺乏统一标准：不同用户用量差异大，需个性化阈值。
- 数据格式不统一：不同供应商的账单格式可能不同，需要预处理。
3. 核心逻辑
1. 数据导入：读取近 12 个月的缴费数据（CSV/Excel）。
2. 数据清洗：处理缺失值、异常值，统一单位。
3. 按月聚合：计算每个用户每月的总用量。
4. 计算月均值：对近 12 个月的数据求平均。
5. 异常检测：当月用量 > 月均值 × 1.2 时，标记为异常。
6. 输出结果：生成预警列表，并可导出为 CSV 或发送邮件/消息。
4. 模块化 Python 代码
项目结构：
utility_anomaly_detector/
├── data/
│ └── usage_data.csv
├── src/
│ ├── data_loader.py
│ ├── data_cleaner.py
│ ├── analyzer.py
│ ├── alert_generator.py
│ └── main.py
├── README.md
└── requirements.txt
"requirements.txt"
pandas>=1.3.0
numpy>=1.21.0
openpyxl>=3.0.0
"src/data_loader.py"
import pandas as pd
def load_data(file_path):
"""
加载 CSV 或 Excel 格式的用量数据
数据格式示例：
user_id, month, water_usage, electricity_usage, gas_usage
001, 2024-01, 10, 200, 5
"""
if file_path.endswith('.csv'):
df = pd.read_csv(file_path)
elif file_path.endswith(('.xls', '.xlsx')):
df = pd.read_excel(file_path)
else:
raise ValueError("Unsupported file format")
return df
"src/data_cleaner.py"
import numpy as np
def clean_data(df):
"""
清洗数据：去除空值，确保用量为数值型
"""
df = df.dropna(subset=['water_usage', 'electricity_usage', 'gas_usage'])
for col in ['water_usage', 'electricity_usage', 'gas_usage']:
df[col] = pd.to_numeric(df[col], errors='coerce')
df = df.dropna()
return df
"src/analyzer.py"
import pandas as pd
def calculate_monthly_avg(df):
"""
计算每个用户每种能源的月平均用量
"""
avg_df = df.groupby('user_id').agg({
'water_usage': 'mean',
'electricity_usage': 'mean',
'gas_usage': 'mean'
}).reset_index()
return avg_df
def detect_anomalies(df, avg_df, threshold_ratio=1.2):
"""
检测异常：当月用量 > 月均值 * threshold_ratio
"""
merged = df.merge(avg_df, on='user_id', suffixes=('', '_avg'))
merged['water_alert'] = merged['water_usage'] > merged['water_usage_avg'] * threshold_ratio
merged['electricity_alert'] = merged['electricity_usage'] > merged['electricity_usage_avg'] * threshold_ratio
merged['gas_alert'] = merged['gas_usage'] > merged['gas_usage_avg'] * threshold_ratio
return merged
"src/alert_generator.py"
def generate_alerts(anomalies_df):
"""
生成预警信息
"""
alerts = []
for _, row in anomalies_df.iterrows():
msg = f"User {row['user_id']} in {row['month']}: "
if row['water_alert']:
msg += f"Water usage {row['water_usage']} > avg {row['water_usage_avg']:.2f}. "
if row['electricity_alert']:
msg += f"Electricity usage {row['electricity_usage']} > avg {row['electricity_usage_avg']:.2f}. "
if row['gas_alert']:
msg += f"Gas usage {row['gas_usage']} > avg {row['gas_usage_avg']:.2f}. "
if any([row['water_alert'], row['electricity_alert'], row['gas_alert']]):
alerts.append(msg.strip())
return alerts
"src/main.py"
from data_loader import load_data
from data_cleaner import clean_data
from analyzer import calculate_monthly_avg, detect_anomalies
from alert_generator import generate_alerts
def main():
# 1. 加载数据
df = load_data('../data/usage_data.csv')
# 2. 清洗数据
df_clean = clean_data(df)
# 3. 计算月均值
avg_df = calculate_monthly_avg(df_clean)
# 4. 检测异常
anomalies = detect_anomalies(df_clean, avg_df, threshold_ratio=1.2)
# 5. 生成预警
alerts = generate_alerts(anomalies)
# 6. 输出结果
print("=== Anomaly Alerts ===")
for alert in alerts:
print(alert)
if __name__ == "__main__":
main()
5. README.md
# 水电燃气异常预警工具
## 功能
- 导入近 12 个月水电燃气缴费数据
- 计算月均用量
- 当月用量超均值 20% 时自动预警
- 输出异常用户及类型
## 安装依赖
bash
pip install -r requirements.txt
## 数据格式
CSV 文件，包含列：
- user_id
- month (YYYY-MM)
- water_usage
- electricity_usage
- gas_usage
## 运行
bash
python src/main.py
## 输出
控制台打印预警信息，可扩展为邮件/短信通知。
6. 使用说明
1. 将近 12 个月的数据整理成
"data/usage_data.csv"。
2. 安装依赖：
"pip install -r requirements.txt"。
3. 运行
"python src/main.py"。
4. 查看控制台输出的预警信息，针对异常用户进行排查。
7. 总结
本工具通过Python + Pandas实现了水电燃气用量的异常检测，具有以下优点：
- 自动化：减少人工检查时间。
- 可扩展：可接入数据库、增加更多能源类型、支持实时数据流。
- 易维护：模块化设计，便于后续优化。
- 业务价值：提前发现浪费隐患，降低运营成本，提升资源利用率。
如果你需要，还可以增加邮件/微信推送功能，或者做成 Web 可视化仪表盘，让物业人员更直观地看到异常数据。
利用AI高效解决实际问题，如果你觉得这个工具好用，欢迎关注我！
)
)
)
)
