3. 数据预处理与清洗
任务说明
本案例针对数字孪生场景中采集到的原始传感器数据,完成三项核心预处理任务:缺失值处理、异常值检测与标准化。输入数据为包含时间戳、温度、湿度和振动幅值的CSV文件,输出为清洗后的标准数据集及处理报告。
数据样例与依赖
使用Python标准库及pandas、numpy、scipy。若未安装,请先执行:pip install pandas numpy scipy。以下为输入数据格式示例(保存为raw_sensor.csv):
timestamp,temp,humidity,vibration
2024-01-01 00:00:00,23.5,45.2,0.12
2024-01-01 00:01:00,,47.1,0.15
2024-01-01 00:02:00,24.1,,0.98
2024-01-01 00:03:00,23.8,46.0,0.13
2024-01-01 00:04:00,99.7,45.8,0.11
2024-01-01 00:05:00,23.6,44.9,0.14
实现代码
import pandas as pd
import numpy as np
from scipy import stats
def preprocess_sensor_data(input_path, output_path):
# 1. 读取原始数据
df = pd.read_csv(input_path, parse_dates=['timestamp'])
report = {'total_rows': len(df)}
# 2. 缺失值处理:数值列用线性插值填充
numeric_cols = ['temp', 'humidity', 'vibration']
df[numeric_cols] = df[numeric_cols].interpolate(method='linear', limit_direction='both')
report['missing_after'] = int(df[numeric_cols].isnull().sum().sum())
# 3. 异常值检测:基于Z-score(阈值3)
z_scores = np.abs(stats.zscore(df[numeric_cols]))
outlier_mask = (z_scores > 3).any(axis=1)
report['outliers_removed'] = int(outlier_mask.sum())
df = df[~outlier_mask].reset_index(drop=True)
# 4. 标准化:Z-score标准化
df_standardized = df.copy()
for col in numeric_cols:
mean = df[col].mean()
std = df[col].std()
df_standardized[col] = (df[col] - mean) / std
# 5. 保存结果
df_standardized.to_csv(output_path, index=False)
# 6. 生成处理报告
report['final_rows'] = len(df_standardized)
report['mean_temp'] = round(df_standardized['temp'].mean(), 6)
report['std_temp'] = round(df_standardized['temp'].std(), 6)
return report
if __name__ == "__main__":
result = preprocess_sensor_data('raw_sensor.csv', 'clean_sensor.csv')
print("处理完成,报告:", result)
执行与验证
将上述代码保存为preprocess.py,与raw_sensor.csv置于同一目录,运行python preprocess.py。验证标准如下:
- 输出文件
clean_sensor.csv存在且行数少于或等于原始行数(因异常值被移除); - 标准化后temp列均值接近0(预期输出:0.0),标准差接近1(预期输出:1.0);
- 控制台报告显示缺失值已填充(missing_after为0),异常值数量与数据中明显偏离值(如99.7)对应。
判定标准
当控制台输出报告中的missing_after为0、outliers_removed大于0,且clean_sensor.csv中temp列均值与标准差分别接近0和1时,本案例实现通过验证。若原始数据无缺失或异常,则对应计数为0,标准化结果仍应满足均值0、标准差1的统计特性。