深入解析scikit-learn在物联网数据分析中的核心应用从数据预处理到模型构建的完整流程与实战案例

引言

随着物联网(IoT)技术的迅速发展，数以亿计的设备不断产生海量数据。这些数据蕴含着巨大的价值，但同时也带来了前所未有的分析挑战。scikit-learn作为Python生态系统中最受欢迎的机器学习库之一，为物联网数据分析提供了强大而灵活的工具集。本文将深入探讨如何利用scikit-learn从原始物联网数据中提取价值，涵盖从数据预处理到模型构建的完整流程，并通过实战案例展示具体应用。

物联网数据特点与挑战

物联网数据具有以下几个显著特点：

海量性：大量设备持续产生数据，形成海量数据集。
高速性：数据生成和传输速度快，需要实时或近实时处理。
多样性：数据类型多样，包括时间序列数据、传感器数据、图像数据等。
噪声与不完整性：由于传感器故障、网络问题等，数据常含有噪声和缺失值。
时空相关性：数据往往具有时间和空间上的相关性。

这些特点给数据分析带来了诸多挑战，如存储问题、计算效率问题、数据质量问题等。scikit-learn提供了一系列工具，帮助我们有效应对这些挑战。

数据预处理

数据预处理是物联网数据分析的关键步骤，直接影响后续模型的性能。下面详细介绍使用scikit-learn进行数据预处理的各个环节。

数据收集与加载

首先，我们需要收集和加载物联网数据。假设我们从一个CSV文件中加载传感器数据：

import pandas as pd import numpy as np # 加载物联网传感器数据 data = pd.read_csv('iot_sensor_data.csv') # 查看数据基本信息 print(data.head()) print(data.info())

数据清洗

物联网数据通常存在缺失值、异常值和噪声，需要进行清洗。

处理缺失值

from sklearn.impute import SimpleImputer # 检查缺失值 print(data.isnull().sum()) # 使用均值填充数值型缺失值 numeric_columns = data.select_dtypes(include=['float64', 'int64']).columns imputer = SimpleImputer(strategy='mean') data[numeric_columns] = imputer.fit_transform(data[numeric_columns]) # 对于分类变量，使用众数填充 categorical_columns = data.select_dtypes(include=['object']).columns imputer = SimpleImputer(strategy='most_frequent') data[categorical_columns] = imputer.fit_transform(data[categorical_columns])

处理异常值

from sklearn.preprocessing import StandardScaler # 使用Z-score方法检测异常值 scaler = StandardScaler() data_scaled = scaler.fit_transform(data[numeric_columns]) data_scaled = pd.DataFrame(data_scaled, columns=numeric_columns) # 定义异常值阈值（通常为±3个标准差） threshold = 3 outliers = (np.abs(data_scaled) > threshold).any(axis=1) # 处理异常值（这里使用中位数替换） for col in numeric_columns: median_val = data.loc[~outliers, col].median() data.loc[outliers, col] = median_val

特征工程

特征工程是提高模型性能的关键步骤，包括特征创建、特征转换和特征选择。

特征创建

对于时间序列物联网数据，我们可以创建一些时间特征：

# 假设数据中有一个时间戳列 data['timestamp'] = pd.to_datetime(data['timestamp']) # 提取时间特征 data['year'] = data['timestamp'].dt.year data['month'] = data['timestamp'].dt.month data['day'] = data['timestamp'].dt.day data['hour'] = data['timestamp'].dt.hour data['dayofweek'] = data['timestamp'].dt.dayofweek # 创建滞后特征（对于时间序列预测） for lag in [1, 2, 3, 24, 48]: # 1小时、2小时、3小时、1天、2天前的值 data[f'temp_lag_{lag}'] = data['temperature'].shift(lag) # 创建滑动窗口统计特征 window_sizes = [6, 12, 24] # 6小时、12小时、24小时窗口 for window in window_sizes: data[f'temp_rolling_mean_{window}'] = data['temperature'].rolling(window=window).mean() data[f'temp_rolling_std_{window}'] = data['temperature'].rolling(window=window).std()

特征转换

from sklearn.preprocessing import StandardScaler, MinMaxScaler, OneHotEncoder, LabelEncoder # 数值特征标准化 scaler = StandardScaler() data[numeric_columns] = scaler.fit_transform(data[numeric_columns]) # 或者使用归一化 # scaler = MinMaxScaler() # data[numeric_columns] = scaler.fit_transform(data[numeric_columns]) # 分类变量编码 # 对于有序分类变量 label_encoder = LabelEncoder() data['device_type_encoded'] = label_encoder.fit_transform(data['device_type']) # 对于无序分类变量，使用独热编码 onehot_encoder = OneHotEncoder(sparse=False) device_status_encoded = onehot_encoder.fit_transform(data[['device_status']]) # 将编码后的特征添加到数据框 encoded_columns = [f'device_status_{i}' for i in range(device_status_encoded.shape[1])] data[encoded_columns] = pd.DataFrame(device_status_encoded, columns=encoded_columns)

特征选择

from sklearn.feature_selection import SelectKBest, f_classif, RFE from sklearn.ensemble import RandomForestClassifier # 假设我们有一个分类任务，X是特征，y是目标变量 X = data.drop(['target', 'timestamp'], axis=1) # 去掉目标变量和时间戳 y = data['target'] # 方法1：使用统计检验选择最佳特征 selector = SelectKBest(score_func=f_classif, k=10) X_new = selector.fit_transform(X, y) selected_features = X.columns[selector.get_support()] print("Selected features:", selected_features) # 方法2：使用递归特征消除 model = RandomForestClassifier(n_estimators=100, random_state=42) rfe = RFE(estimator=model, n_features_to_select=10) X_rfe = rfe.fit_transform(X, y) selected_features_rfe = X.columns[rfe.support_] print("RFE selected features:", selected_features_rfe) # 方法3：基于特征重要性 model.fit(X, y) importances = model.feature_importances_ indices = np.argsort(importances)[::-1] top_features = X.columns[indices[:10]] print("Top features by importance:", top_features)

数据降维

当特征数量过多时，可以使用降维技术减少特征数量，同时保留大部分信息。

from sklearn.decomposition import PCA # 主成分分析 pca = PCA(n_components=0.95) # 保留95%的方差 X_pca = pca.fit_transform(X) print("Original number of features:", X.shape[1]) print("Reduced number of features:", X_pca.shape[1])

模型构建

完成数据预处理后，我们可以开始构建机器学习模型。根据物联网数据分析的不同任务，可以选择不同的模型。

监督学习在IoT中的应用

监督学习适用于有标签数据的场景，如预测性维护、异常检测等。

分类模型

from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV from sklearn.metrics import classification_report, confusion_matrix, accuracy_score from sklearn.linear_model import LogisticRegression from sklearn.svm import SVC from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier from sklearn.neural_network import MLPClassifier # 划分训练集和测试集 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) # 定义多个分类模型 models = { 'Logistic Regression': LogisticRegression(max_iter=1000), 'SVM': SVC(), 'Random Forest': RandomForestClassifier(n_estimators=100), 'Gradient Boosting': GradientBoostingClassifier(n_estimators=100), 'Neural Network': MLPClassifier(hidden_layer_sizes=(100, 50), max_iter=500) } # 评估各模型 for name, model in models.items(): # 训练模型 model.fit(X_train, y_train) # 预测 y_pred = model.predict(X_test) # 评估 print(f"n{name}:") print(f"Accuracy: {accuracy_score(y_test, y_pred):.4f}") print(classification_report(y_test, y_pred)) # 交叉验证 cv_scores = cross_val_score(model, X, y, cv=5) print(f"Cross-validation scores: {cv_scores}") print(f"Mean CV accuracy: {cv_scores.mean():.4f}")

回归模型

对于连续值预测任务，如温度预测、能耗预测等：

from sklearn.linear_model import LinearRegression, Ridge, Lasso from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor from sklearn.svm import SVR from sklearn.metrics import mean_squared_error, r2_score, mean_absolute_error # 假设y是连续值 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) # 定义回归模型 regressors = { 'Linear Regression': LinearRegression(), 'Ridge Regression': Ridge(alpha=1.0), 'Lasso Regression': Lasso(alpha=1.0), 'Random Forest': RandomForestRegressor(n_estimators=100), 'Gradient Boosting': GradientBoostingRegressor(n_estimators=100), 'SVR': SVR(kernel='rbf') } # 评估各回归模型 for name, model in regressors.items(): # 训练模型 model.fit(X_train, y_train) # 预测 y_pred = model.predict(X_test) # 评估 print(f"n{name}:") print(f"MAE: {mean_absolute_error(y_test, y_pred):.4f}") print(f"MSE: {mean_squared_error(y_test, y_pred):.4f}") print(f"RMSE: {np.sqrt(mean_squared_error(y_test, y_pred)):.4f}") print(f"R²: {r2_score(y_test, y_pred):.4f}")

超参数调优

# 以随机森林为例进行超参数调优 param_grid = { 'n_estimators': [50, 100, 200], 'max_depth': [None, 10, 20, 30], 'min_samples_split': [2, 5, 10], 'min_samples_leaf': [1, 2, 4] } rf = RandomForestClassifier(random_state=42) grid_search = GridSearchCV(estimator=rf, param_grid=param_grid, cv=5, n_jobs=-1, verbose=2) grid_search.fit(X_train, y_train) # 最佳参数和模型 print(f"Best parameters: {grid_search.best_params_}") best_rf = grid_search.best_estimator_ # 使用最佳模型进行预测 y_pred = best_rf.predict(X_test) print(f"Optimized model accuracy: {accuracy_score(y_test, y_pred):.4f}")

无监督学习在IoT中的应用

无监督学习适用于没有标签数据的场景，如聚类、异常检测等。

聚类分析

from sklearn.cluster import KMeans, DBSCAN, AgglomerativeClustering from sklearn.metrics import silhouette_score import matplotlib.pyplot as plt # K-means聚类 # 确定最佳聚类数 silhouette_scores = [] for k in range(2, 11): kmeans = KMeans(n_clusters=k, random_state=42) cluster_labels = kmeans.fit_predict(X) silhouette_avg = silhouette_score(X, cluster_labels) silhouette_scores.append(silhouette_avg) print(f"For n_clusters = {k}, the average silhouette_score is: {silhouette_avg:.4f}") # 绘制轮廓系数图 plt.figure(figsize=(10, 6)) plt.plot(range(2, 11), silhouette_scores, marker='o') plt.xlabel('Number of clusters') plt.ylabel('Silhouette Score') plt.title('Silhouette Score for Optimal k') plt.show() # 使用最佳聚类数进行K-means聚类 optimal_k = np.argmax(silhouette_scores) + 2 # +2因为range从2开始 kmeans = KMeans(n_clusters=optimal_k, random_state=42) cluster_labels = kmeans.fit_predict(X) data['cluster'] = cluster_labels # DBSCAN聚类 dbscan = DBSCAN(eps=0.5, min_samples=5) dbscan_labels = dbscan.fit_predict(X) data['dbscan_cluster'] = dbscan_labels # 层次聚类 agg = AgglomerativeClustering(n_clusters=optimal_k) agg_labels = agg.fit_predict(X) data['agg_cluster'] = agg_labels

异常检测

from sklearn.ensemble import IsolationForest from sklearn.svm import OneClassSVM from sklearn.neighbors import LocalOutlierFactor # Isolation Forest iso_forest = IsolationForest(contamination=0.05, random_state=42) iso_labels = iso_forest.fit_predict(X) # 转换标签：-1表示异常，1表示正常 data['iso_anomaly'] = iso_labels # One-Class SVM oc_svm = OneClassSVM(nu=0.05) svm_labels = oc_svm.fit_predict(X) data['svm_anomaly'] = svm_labels # Local Outlier Factor lof = LocalOutlierFactor(n_neighbors=20, contamination=0.05) lof_labels = lof.fit_predict(X) data['lof_anomaly'] = lof_labels # 统计异常检测结果 print("Isolation Forest anomalies:", sum(iso_labels == -1)) print("One-Class SVM anomalies:", sum(svm_labels == -1)) print("LOF anomalies:", sum(lof_labels == -1))

时间序列分析

物联网数据通常是时间序列数据，可以使用专门的时间序列分析方法：

from sklearn.linear_model import LinearRegression from sklearn.multioutput import MultiOutputRegressor # 创建时间序列特征 def create_timeseries_features(data, target_col, window_size=5): X, y = [], [] for i in range(len(data) - window_size): X.append(data[i:i+window_size].drop(target_col, axis=1).values.flatten()) y.append(data.iloc[i+window_size][target_col]) return np.array(X), np.array(y) # 假设我们有一个时间序列数据集ts_data ts_X, ts_y = create_timeseries_features(data, 'temperature', window_size=24) # 使用24小时窗口 # 划分训练集和测试集 ts_X_train, ts_X_test, ts_y_train, ts_y_test = train_test_split(ts_X, ts_y, test_size=0.2, random_state=42) # 使用线性回归进行预测 model = LinearRegression() model.fit(ts_X_train, ts_y_train) ts_y_pred = model.predict(ts_X_test) # 评估 print(f"Time Series Prediction RMSE: {np.sqrt(mean_squared_error(ts_y_test, ts_y_pred)):.4f}") print(f"Time Series Prediction R²: {r2_score(ts_y_test, ts_y_pred):.4f}") # 可视化预测结果 plt.figure(figsize=(15, 6)) plt.plot(ts_y_test, label='Actual') plt.plot(ts_y_pred, label='Predicted') plt.legend() plt.title('Time Series Prediction') plt.show()

实战案例：智能建筑能耗预测与异常检测

为了更好地理解scikit-learn在物联网数据分析中的应用，我们通过一个智能建筑能耗预测与异常检测的实战案例来展示完整流程。

案例背景与数据集介绍

假设我们有一个智能建筑的物联网数据集，包含以下信息：

时间戳
温度、湿度、光照等环境传感器数据
电力、水、气等能耗数据
建筑占用情况（人数、活动区域等）
HVAC系统运行状态

我们的目标是：

预测未来24小时的电力消耗
检测能耗异常情况

数据预处理步骤详解

首先，我们加载和预处理数据：

import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler, MinMaxScaler, OneHotEncoder from sklearn.impute import SimpleImputer from sklearn.compose import ColumnTransformer from sklearn.pipeline import Pipeline from sklearn.model_selection import train_test_split # 加载数据 data = pd.read_csv('smart_building_data.csv') # 转换时间戳 data['timestamp'] = pd.to_datetime(data['timestamp']) data.set_index('timestamp', inplace=True) # 检查数据质量 print(data.info()) print(data.describe()) # 处理缺失值 # 数值列使用均值填充 numeric_cols = data.select_dtypes(include=['float64', 'int64']).columns data[numeric_cols] = data[numeric_cols].fillna(data[numeric_cols].mean()) # 分类列使用众数填充 categorical_cols = data.select_dtypes(include=['object']).columns for col in categorical_cols: data[col].fillna(data[col].mode()[0], inplace=True) # 创建时间特征 data['hour'] = data.index.hour data['dayofweek'] = data.index.dayofweek data['month'] = data.index.month data['is_weekend'] = (data.index.dayofweek >= 5).astype(int) # 创建滞后特征（用于能耗预测） for lag in [1, 2, 3, 24, 48, 168]: # 1小时、2小时、3小时、1天、2天、1周前的值 data[f'power_lag_{lag}'] = data['power_consumption'].shift(lag) # 创建滑动窗口统计特征 window_sizes = [6, 12, 24, 168] # 6小时、12小时、1天、1周 for window in window_sizes: data[f'power_rolling_mean_{window}'] = data['power_consumption'].rolling(window=window).mean() data[f'power_rolling_std_{window}'] = data['power_consumption'].rolling(window=window).std() # 删除包含NaN的行（由于创建滞后特征和滑动窗口特征） data.dropna(inplace=True) # 定义特征和目标变量 X = data.drop('power_consumption', axis=1) y = data['power_consumption'] # 区分数值列和分类列 numeric_features = X.select_dtypes(include=['float64', 'int64']).columns categorical_features = X.select_dtypes(include=['object']).columns # 创建预处理管道 numeric_transformer = Pipeline(steps=[ ('scaler', StandardScaler()) ]) categorical_transformer = Pipeline(steps=[ ('onehot', OneHotEncoder(handle_unknown='ignore')) ]) preprocessor = ColumnTransformer( transformers=[ ('num', numeric_transformer, numeric_features), ('cat', categorical_transformer, categorical_features) ]) # 划分训练集和测试集 X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

模型构建与评估

能耗预测模型

from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor from sklearn.linear_model import LinearRegression from sklearn.metrics import mean_squared_error, r2_score, mean_absolute_error from sklearn.model_selection import cross_val_score import matplotlib.pyplot as plt # 定义模型 models = { 'Linear Regression': LinearRegression(), 'Random Forest': RandomForestRegressor(n_estimators=100, random_state=42), 'Gradient Boosting': GradientBoostingRegressor(n_estimators=100, random_state=42) } # 创建完整的处理和建模管道 results = {} for name, model in models.items(): # 创建管道 pipeline = Pipeline(steps=[('preprocessor', preprocessor), ('model', model)]) # 训练模型 pipeline.fit(X_train, y_train) # 预测 y_pred = pipeline.predict(X_test) # 评估 mse = mean_squared_error(y_test, y_pred) rmse = np.sqrt(mse) mae = mean_absolute_error(y_test, y_pred) r2 = r2_score(y_test, y_pred) # 交叉验证 cv_scores = cross_val_score(pipeline, X, y, cv=5, scoring='neg_mean_squared_error') cv_rmse = np.sqrt(-cv_scores.mean()) results[name] = { 'RMSE': rmse, 'MAE': mae, 'R²': r2, 'CV RMSE': cv_rmse, 'model': pipeline } print(f"n{name}:") print(f"RMSE: {rmse:.4f}") print(f"MAE: {mae:.4f}") print(f"R²: {r2:.4f}") print(f"Cross-validated RMSE: {cv_rmse:.4f}") # 选择最佳模型 best_model_name = min(results, key=lambda k: results[k]['RMSE']) best_model = results[best_model_name]['model'] print(f"nBest model: {best_model_name}") # 可视化预测结果 y_pred = best_model.predict(X_test) plt.figure(figsize=(15, 6)) plt.plot(y_test.values, label='Actual') plt.plot(y_pred, label='Predicted') plt.legend() plt.title(f'Energy Consumption Prediction using {best_model_name}') plt.show() # 特征重要性分析（仅对树模型有效） if best_model_name in ['Random Forest', 'Gradient Boosting']: # 获取特征名称 feature_names = [] for name, transformer, columns in preprocessor.transformers_: if name == 'cat': # 对于分类特征，获取独热编码后的特征名 cat_features = transformer.named_steps['onehot'].get_feature_names_out(columns) feature_names.extend(cat_features) else: feature_names.extend(columns) # 获取特征重要性 importances = best_model.named_steps['model'].feature_importances_ indices = np.argsort(importances)[::-1] # 打印特征重要性 print("nFeature importance:") for i in range(min(20, len(feature_names))): print(f"{feature_names[indices[i]]}: {importances[indices[i]]:.4f}") # 可视化特征重要性 plt.figure(figsize=(12, 8)) plt.title('Feature Importance') plt.bar(range(min(20, len(importances))), importances[indices[:20]], align='center') plt.xticks(range(min(20, len(importances))), [feature_names[i] for i in indices[:20]], rotation=90) plt.tight_layout() plt.show()

异常检测模型

from sklearn.ensemble import IsolationForest from sklearn.svm import OneClassSVM from sklearn.neighbors import LocalOutlierFactor # 使用预处理后的数据进行异常检测 X_processed = preprocessor.fit_transform(X) # Isolation Forest iso_forest = IsolationForest(contamination=0.05, random_state=42) iso_labels = iso_forest.fit_predict(X_processed) iso_scores = iso_forest.decision_function(X_processed) # One-Class SVM oc_svm = OneClassSVM(nu=0.05) svm_labels = oc_svm.fit_predict(X_processed) svm_scores = oc_svm.decision_function(X_processed) # Local Outlier Factor lof = LocalOutlierFactor(n_neighbors=20, contamination=0.05) lof_labels = lof.fit_predict(X_processed) lof_scores = lof.negative_outlier_factor_ # 将异常检测结果添加到原始数据 data['iso_anomaly'] = iso_labels data['iso_score'] = iso_scores data['svm_anomaly'] = svm_labels data['svm_score'] = svm_scores data['lof_anomaly'] = lof_labels data['lof_score'] = lof_scores # 统计异常检测结果 print("nAnomaly Detection Results:") print(f"Isolation Forest: {sum(iso_labels == -1)} anomalies detected ({sum(iso_labels == -1)/len(iso_labels)*100:.2f}%)") print(f"One-Class SVM: {sum(svm_labels == -1)} anomalies detected ({sum(svm_labels == -1)/len(svm_labels)*100:.2f}%)") print(f"LOF: {sum(lof_labels == -1)} anomalies detected ({sum(lof_labels == -1)/len(lof_labels)*100:.2f}%)") # 可视化异常检测结果 plt.figure(figsize=(15, 10)) # 能耗时间序列与异常点 plt.subplot(3, 1, 1) plt.plot(data.index, data['power_consumption'], label='Power Consumption') anomalies = data[data['iso_anomaly'] == -1] plt.scatter(anomalies.index, anomalies['power_consumption'], color='red', label='Anomalies') plt.title('Power Consumption with Anomalies (Isolation Forest)') plt.legend() # 异常分数分布 plt.subplot(3, 1, 2) plt.hist(data['iso_score'], bins=50) plt.title('Anomaly Score Distribution (Isolation Forest)') # 按时间统计异常数量 plt.subplot(3, 1, 3) daily_anomalies = data['iso_anomaly'].resample('D').apply(lambda x: sum(x == -1)) plt.plot(daily_anomalies.index, daily_anomalies.values) plt.title('Daily Anomaly Count') plt.tight_layout() plt.show() # 分析异常情况 anomalies = data[data['iso_anomaly'] == -1] print("nAnomaly Analysis:") print(f"Average power consumption during anomalies: {anomalies['power_consumption'].mean():.2f}") print(f"Average power consumption during normal periods: {data[data['iso_anomaly'] == 1]['power_consumption'].mean():.2f}") print(f"Anomaly increase: {(anomalies['power_consumption'].mean() / data[data['iso_anomaly'] == 1]['power_consumption'].mean() - 1) * 100:.2f}%") # 按小时分析异常分布 hourly_anomaly_count = anomalies.groupby(anomalies.index.hour).size() hourly_total_count = data.groupby(data.index.hour).size() hourly_anomaly_rate = hourly_anomaly_count / hourly_total_count * 100 plt.figure(figsize=(12, 6)) plt.bar(hourly_anomaly_rate.index, hourly_anomaly_rate.values) plt.title('Hourly Anomaly Rate') plt.xlabel('Hour of Day') plt.ylabel('Anomaly Rate (%)') plt.show()

结果分析与优化

模型优化

from sklearn.model_selection import GridSearchCV # 对最佳模型进行超参数调优 if best_model_name == 'Random Forest': param_grid = { 'model__n_estimators': [50, 100, 200], 'model__max_depth': [None, 10, 20, 30], 'model__min_samples_split': [2, 5, 10], 'model__min_samples_leaf': [1, 2, 4] } elif best_model_name == 'Gradient Boosting': param_grid = { 'model__n_estimators': [50, 100, 200], 'model__learning_rate': [0.01, 0.1, 0.2], 'model__max_depth': [3, 5, 7], 'model__min_samples_split': [2, 5, 10] } else: # Linear Regression print("Linear Regression does not have hyperparameters to tune.") param_grid = {} if param_grid: # 创建管道 pipeline = Pipeline(steps=[('preprocessor', preprocessor), ('model', models[best_model_name])]) # 网格搜索 grid_search = GridSearchCV(pipeline, param_grid, cv=5, scoring='neg_mean_squared_error', n_jobs=-1, verbose=1) grid_search.fit(X_train, y_train) # 最佳参数和模型 print(f"nBest parameters: {grid_search.best_params_}") best_model_optimized = grid_search.best_estimator_ # 评估优化后的模型 y_pred_optimized = best_model_optimized.predict(X_test) rmse_optimized = np.sqrt(mean_squared_error(y_test, y_pred_optimized)) r2_optimized = r2_score(y_test, y_pred_optimized) print(f"nOptimized {best_model_name}:") print(f"RMSE: {rmse_optimized:.4f}") print(f"R²: {r2_optimized:.4f}") # 比较优化前后的性能 print(f"nImprovement:") print(f"RMSE reduced by: {(results[best_model_name]['RMSE'] - rmse_optimized):.4f}") print(f"R² improved by: {(r2_optimized - results[best_model_name]['R²']):.4f}")

部署与监控

import joblib import datetime # 保存模型 joblib.dump(best_model_optimized, 'energy_consumption_model.pkl') joblib.dump(preprocessor, 'data_preprocessor.pkl') joblib.dump(iso_forest, 'anomaly_detector.pkl') # 模拟模型部署和监控 def predict_energy_consumption(new_data): """ 使用训练好的模型预测能耗 """ # 加载模型和预处理器 model = joblib.load('energy_consumption_model.pkl') # 预处理新数据 # 注意：新数据需要与训练数据具有相同的特征 prediction = model.predict(new_data) return prediction def detect_anomalies(new_data): """ 使用训练好的模型检测异常 """ # 加载模型和预处理器 preprocessor = joblib.load('data_preprocessor.pkl') anomaly_detector = joblib.load('anomaly_detector.pkl') # 预处理新数据 processed_data = preprocessor.transform(new_data) # 检测异常 anomalies = anomaly_detector.predict(processed_data) anomaly_scores = anomaly_detector.decision_function(processed_data) return anomalies, anomaly_scores # 模拟实时数据流监控 print("nSimulating real-time monitoring...") # 生成一些模拟数据 np.random.seed(42) n_samples = 24 # 24小时数据 simulated_data = pd.DataFrame({ 'temperature': np.random.normal(22, 2, n_samples), 'humidity': np.random.normal(45, 5, n_samples), 'occupancy': np.random.randint(0, 100, n_samples), 'hour': range(n_samples), 'dayofweek': [1] * n_samples, # 假设都是周二 'month': [6] * n_samples, # 假设都是六月 'is_weekend': [0] * n_samples, # 工作日 }) # 添加一些滞后特征（模拟） for lag in [1, 2, 3, 24, 48, 168]: simulated_data[f'power_lag_{lag}'] = np.random.normal(500, 50, n_samples) # 添加滑动窗口统计特征（模拟） for window in [6, 12, 24, 168]: simulated_data[f'power_rolling_mean_{window}'] = np.random.normal(500, 50, n_samples) simulated_data[f'power_rolling_std_{window}'] = np.random.normal(50, 5, n_samples) # 添加一个分类特征 simulated_data['hvac_status'] = np.random.choice(['on', 'off', 'standby'], n_samples) # 确保列顺序与训练数据相同 # 注意：在实际应用中，需要确保新数据的特征与训练数据完全一致 # 这里我们假设simulated_data已经包含了所有必要的特征 # 预测能耗 predictions = predict_energy_consumption(simulated_data) print("nEnergy consumption predictions for the next 24 hours:") for i, pred in enumerate(predictions): print(f"Hour {i+1}: {pred:.2f} kWh") # 检测异常 anomalies, scores = detect_anomalies(simulated_data) print("nAnomaly detection results:") for i, (anomaly, score) in enumerate(zip(anomalies, scores)): status = "ANOMALY" if anomaly == -1 else "Normal" print(f"Hour {i+1}: {status} (Score: {score:.4f})") # 模拟模型性能监控 print("nSimulating model performance monitoring...") # 假设我们有一些新的真实数据来评估模型 # 在实际应用中，这可能是定期收集的数据 new_real_data = pd.read_csv('new_energy_data.csv') # 假设我们有新数据 new_X = new_real_data.drop('power_consumption', axis=1) new_y = new_real_data['power_consumption'] # 使用模型进行预测 new_predictions = best_model_optimized.predict(new_X) # 计算性能指标 new_rmse = np.sqrt(mean_squared_error(new_y, new_predictions)) new_r2 = r2_score(new_y, new_predictions) print(f"nModel performance on new data:") print(f"RMSE: {new_rmse:.4f}") print(f"R²: {new_r2:.4f}") # 比较与原始测试集的性能 print(f"nPerformance comparison:") print(f"Original test RMSE: {results[best_model_name]['RMSE']:.4f}") print(f"New data RMSE: {new_rmse:.4f}") print(f"Performance change: {((new_rmse - results[best_model_name]['RMSE']) / results[best_model_name]['RMSE'] * 100):.2f}%") # 如果性能下降超过阈值，触发模型重新训练 if new_rmse > results[best_model_name]['RMSE'] * 1.1: # 10%的性能下降 print("nWarning: Model performance has degraded significantly. Consider retraining the model.") # 模拟模型重新训练 print("nRetraining model with new data...") # 合并原始训练数据和新数据 combined_X = pd.concat([X_train, new_X]) combined_y = pd.concat([y_train, new_y]) # 重新训练模型 best_model_optimized.fit(combined_X, combined_y) # 评估重新训练后的模型 retrained_predictions = best_model_optimized.predict(X_test) retrained_rmse = np.sqrt(mean_squared_error(y_test, retrained_predictions)) retrained_r2 = r2_score(y_test, retrained_predictions) print(f"nRetrained model performance:") print(f"RMSE: {retrained_rmse:.4f}") print(f"R²: {retrained_r2:.4f}") # 保存重新训练的模型 joblib.dump(best_model_optimized, 'energy_consumption_model_retrained.pkl') print("nRetrained model saved.")