基于支持向量回归（SVR）的空气质量预测

基于支持向量回归（SVR）的空气质量预测

1.作者介绍

王浩，男，西安工程大学电子信息学院，2024级研究生
研究方向：图像法识别羊绒羊毛
电子邮件：3122496059@qq.com

王晓睿，男，西安工程大学电子信息学院，2024级研究生，张宏伟人工智能课题组
研究方向：智能视觉检测与工业自动化技术
电子邮件：3234002295@qq.com

2.支持向量回归（SVR）算法介绍

2.1 算法原理

SVR（Support Vector Regression）是支持向量机（SVM）在回归问题上的应用，SVR通过最小化模型复杂度（‖w‖²）和控制ε-insensitive误差，在拟合精度与泛化能力之间取得平衡。其核心是构建一个宽度为2ε的"回归管道"，仅对落在管道外的样本计算损失，通过支持向量确定最优回归超平面。

2.2 关键概念

（1）ε-不敏感损失：只有当预测值与真实值的偏差超过阈值 ε 时才计算误差。
（2）间隔带（ε-tube）：由超平面和两侧边界（±ε）构成的区域，模型允许误差在此范围内。
（3）支持向量：位于间隔带外或边界上的点，决定模型形状。

2.3算法特点

（1）容忍机制：采用ε-不敏感损失函数，允许预测值在±ε范围内自由波动而不受惩罚，鲁棒性强，抗噪声数据。
（2）稀疏解特性：最终模型仅依赖少量支持向量（管道边界外的样本），计算高效，适合小样本，高维数据。
（3）核技巧兼容：通过核函数隐式映射到高维空间，可处理复杂非线性关系。

2.4与其他回归方法对比

3.基于支持向量回归（SVR）的空气质量预测实验

3.1数据集介绍

数据来自北京市环境监测中心 ，包含12个国家级空气质量监测站点的每小时数据。监测的污染物包括6种主要空气污染物（如PM2.5、PM10、SO₂等）和6项相关气象变量（如温度、湿度、风速等）。
下载链接为：https://archive.ics.uci.edu/dataset/501/beijing+multi+site+air+quality+data

本次实验使用天坛检测站点的数据：

3.2代码实现

（1）导入必要的库

（2）初始化配置

（3）数据加载和预处理

（4）模型训练

3.3完整代码

-*- coding: utf-8 -*-
"""
基于SVR的空气质量预测系统
数据集：空气质量监测数据（每小时记录）
功能：预测PM2.5浓度并生成专业分析报告
"""# 1. 导入必要库
import os
import pandas as pd
import numpy as np
from sklearn.svm import SVR
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.model_selection import train_test_split, RandomizedSearchCV
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from scipy.stats import loguniform
import matplotlib.pyplot as plt
from matplotlib import rcParams
import joblib# 2. 全局配置
def set_global_config():"""配置全局参数"""rcParams['font.sans-serif'] = ['Microsoft YaHei']  # 中文字体rcParams['axes.unicode_minus'] = False  # 显示负号plt.style.use('ggplot')  # 美观的绘图样式pd.set_option('display.max_columns', 20)  # 显示更多列# 3. 数据加载与预处理
class DataProcessor:"""数据处理管道"""def __init__(self, file_path):self.file_path = file_pathself.df = Noneself.numeric_features = []self.categorical_features = []def load_data(self):"""加载并清洗数据"""try:# 尝试不同编码方式读取文件try:self.df = pd.read_csv(self.file_path, encoding='utf-8')except:self.df = pd.read_csv(self.file_path, encoding='gbk')print(f"✅ 成功加载数据，原始记录数: {len(self.df)}")print("可用列名:", self.df.columns.tolist())# 清洗列名：去除前后空格、转换为小写self.df.columns = self.df.columns.str.strip().str.lower()# 检查必要的列是否存在required_cols = ['pm2.5', 'pm10', 'temp', 'pres']missing_cols = [col for col in required_cols if col not in self.df.columns]if missing_cols:raise ValueError(f"缺失必要列: {missing_cols}")# 处理缺失值self._handle_missing_values()return self.dfexcept Exception as e:print(f"❌ 数据加载失败: {str(e)}")exit()def _handle_missing_values(self):"""处理缺失值"""# 统一缺失值表示self.df = self.df.replace(['NA', 'N/A', 'na', 'n/a', 'NULL', 'null'], np.nan)# 删除关键列缺失的行initial_count = len(self.df)self.df = self.df.dropna(subset=['pm2.5'])  # 只删除PM2.5缺失的行print(f"✅ 缺失值处理后记录数: {len(self.df)} (删除{initial_count-len(self.df)}条)")# 转换数据类型self._convert_dtypes()def _convert_dtypes(self):"""类型转换"""numeric_cols = ['year', 'month', 'day', 'hour', 'pm2.5', 'pm10','so2', 'no2', 'co', 'o3', 'temp', 'pres', 'dew', 'rain', 'wspm']# 只转换实际存在的列numeric_cols = [col for col in numeric_cols if col in self.df.columns]self.df[numeric_cols] = self.df[numeric_cols].apply(pd.to_numeric, errors='coerce')def feature_engineering(self):"""高级特征工程"""# 时间周期特征self.df['month_sin'] = np.sin(2 * np.pi * self.df['month']/12)self.df['month_cos'] = np.cos(2 * np.pi * self.df['month']/12)self.df['hour_sin'] = np.sin(2 * np.pi * self.df['hour']/24)self.df['hour_cos'] = np.cos(2 * np.pi * self.df['hour']/24)# 气象交互特征（使用实际存在的列）if 'dew' in self.df.columns:self.df['dew_point_diff'] = self.df['temp'] - self.df['dew']else:print("⚠️ 未找到'dew'列，跳过露点差特征")# 定义特征列（根据实际存在的列调整）base_features = ['month_sin', 'month_cos', 'hour_sin', 'hour_cos','pm10', 'so2', 'no2', 'co', 'temp', 'pres']# 添加露点差特征（如果存在）if 'dew_point_diff' in self.df.columns:base_features.append('dew_point_diff')self.numeric_features = [f for f in base_features if f in self.df.columns]self.categorical_features = ['wd'] if 'wd' in self.df.columns else []print(f"🔧 使用的数值特征: {self.numeric_features}")print(f"🔧 使用的分类特征: {self.categorical_features}")return self.df# 4. 模型训练类
class SVRTrainer:"""SVR模型训练管道"""def __init__(self, numeric_features, categorical_features):self.numeric_features = numeric_featuresself.categorical_features = categorical_featuresself.preprocessor = self._build_preprocessor()self.model = Nonedef _build_preprocessor(self):"""构建特征预处理管道（包含缺失值处理）"""numeric_transformer = Pipeline(steps=[('imputer', SimpleImputer(strategy='median')),  # 中位数填充('scaler', StandardScaler())])categorical_transformer = Pipeline(steps=[('imputer', SimpleImputer(strategy='most_frequent')),  # 众数填充('onehot', OneHotEncoder(handle_unknown='ignore'))])transformers = []if self.numeric_features:transformers.append(('num', numeric_transformer, self.numeric_features))if self.categorical_features:transformers.append(('cat', categorical_transformer, self.categorical_features))return ColumnTransformer(transformers=transformers)def train(self, X_train, y_train):"""模型训练流程"""param_dist = {'kernel': ['rbf', 'linear'],'C': loguniform(1e-1, 1e2),'epsilon': [0.01, 0.1, 0.5],'gamma': ['scale', 'auto']}svr = SVR()search = RandomizedSearchCV(svr, param_dist, n_iter=15, cv=3,scoring='neg_mean_squared_error', n_jobs=-1, random_state=42, verbose=1)print("⏳ 开始模型训练...")X_processed = self.preprocessor.fit_transform(X_train)search.fit(X_processed, y_train)self.model = search.best_estimator_print(f"✅ 训练完成，最佳参数: {search.best_params_}")return self.modeldef evaluate(self, X_test, y_test):"""模型评估"""X_processed = self.preprocessor.transform(X_test)y_pred = self.model.predict(X_processed)metrics = {'MAE': mean_absolute_error(y_test, y_pred),'RMSE': np.sqrt(mean_squared_error(y_test, y_pred)),'R²': r2_score(y_test, y_pred)}return metrics, y_pred# 5. 可视化模块
class ResultVisualizer:"""结果可视化生成器"""@staticmethoddef plot_comparison(y_true, y_pred, save_path):"""预测对比图"""plt.figure(figsize=(14, 6))plt.plot(y_true[:100], label='真实值', color='#2c7bb6', marker='o')plt.plot(y_pred[:100], label='预测值', color='#d7191c', linestyle='--')plt.title('PM2.5浓度预测对比（前100样本）', fontsize=14)plt.xlabel('样本序号')plt.ylabel('PM2.5浓度 (μg/m³)')plt.legend()plt.grid(True, alpha=0.3)plt.savefig(save_path, dpi=300, bbox_inches='tight')plt.close()print(f"📈 预测对比图已保存至: {save_path}")@staticmethoddef plot_residuals(y_true, y_pred, save_path):"""残差分析图"""residuals = y_true - y_predplt.figure(figsize=(10, 6))plt.scatter(y_pred, residuals, alpha=0.6, c=np.abs(residuals), cmap='viridis')plt.colorbar(label='残差绝对值')plt.axhline(0, color='red', linestyle='--')plt.title('预测残差分布', fontsize=14)plt.xlabel('预测值 (μg/m³)')plt.ylabel('残差')plt.grid(True, alpha=0.3)plt.savefig(save_path, dpi=300, bbox_inches='tight')plt.close()print(f"📊 残差分析图已保存至: {save_path}")# 6. 主程序
def main():set_global_config()# 初始化路径current_dir = os.path.dirname(os.path.abspath(__file__))data_path = os.path.join(current_dir, 'air_quality_data.csv')		#数据集文件路径output_dir = os.path.join(current_dir, 'results')                   #实验结果保存结果路径os.makedirs(output_dir, exist_ok=True)# 数据预处理processor = DataProcessor(data_path)df = processor.load_data()df = processor.feature_engineering()# 检查数据中是否还有NaNprint("\n🔍 数据缺失值检查:")print(df[processor.numeric_features + processor.categorical_features].isna().sum())# 划分数据集X = df[processor.numeric_features + processor.categorical_features]y = df['pm2.5']  # 目标变量X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)# 模型训练trainer = SVRTrainer(processor.numeric_features, processor.categorical_features)model = trainer.train(X_train, y_train)# 模型评估metrics, y_pred = trainer.evaluate(X_test, y_test)print("\n📊 模型性能评估:")for k, v in metrics.items():print(f"{k}: {v:.4f}")# 保存结果results = pd.DataFrame({'真实值': y_test.values,'预测值': y_pred,'残差': y_test.values - y_pred})# 添加日期信息（如果存在）date_cols = ['year', 'month', 'day']if all(col in df.columns for col in date_cols):results['日期'] = df.loc[y_test.index, date_cols].apply(lambda x: f"{int(x[0])}-{int(x[1])}-{int(x[2])}", axis=1)results_path = os.path.join(output_dir, 'prediction_results.csv')results.to_csv(results_path, index=False, encoding='utf_8_sig')print(f"💾 预测结果已保存至: {results_path}")# 可视化ResultVisualizer.plot_comparison(y_test.values, y_pred, os.path.join(output_dir, 'comparison.png'))ResultVisualizer.plot_residuals(y_test.values, y_pred, os.path.join(output_dir, 'residuals.png'))# 保存模型model_path = os.path.join(output_dir, 'svr_model.pkl')joblib.dump({'model': model, 'preprocessor': trainer.preprocessor}, model_path)print(f"🤖 模型已保存至: {model_path}")if __name__ == "__main__":
main()

百度网盘链接：基于支持向量回归（SVR）的空气质量预测
链接: https://pan.baidu.com/s/1JxU1SCobvCW_cyhPQYvLnw 提取码: qy7k

3.4结果展示

（1）预测对比图

（2）残差分布图