企业官网建设流程全解析

简介：这是一套面向计算机专业本科生的高分毕业设计级空气质量监测与预测系统源码，融合LSTM时序建模能力与Django Web工程实践，解决环境数据实时可视化、历史趋势分析及未来PM2.5等指标短期预测等典型课题需求，特别适合作为期末大作业、课程设计或毕设选题的完整技术落地方案。资源包共260个文件，含15个核心Python脚本（涵盖LSTM训练、数据预处理与API接口）、8个HTML前端页面（如current.html、analysis.html等实现动态展示）、23个JS交互逻辑、164个SCSS样式文件保障界面可维护性，以及CSV实测数据（t_pm25.csv、pm25.csv）和SQLite3本地数据库，整体7.07MB，结构清晰、注释详尽，新手可快速理解模块分工并完成本地部署。目前已有384人学习下载，项目获导师高度认可，评分98分，提供从数据采集、模型训练、Web服务发布到多维度可视化的一站式实现，附带省份分布图（provinces.html）与实时监测页，具备直接复用与二次开发价值。

1. 为什么用 LSTM + Django 做空气质量预测，不是“炫技”，而是真能跑通的最小闭环

你手上有近一年的 hourly 空气质量数据（PM2.5、PM10、SO₂、NO₂、O₃、CO、温度、湿度、风速），想做一个能在线查看历史趋势、实时更新、还能未来 24 小时逐小时预测的系统——不是 Jupyter Notebook 里跑通就完事，而是部署到公司内网服务器、让环保科同事用浏览器打开就能查、能导出、能设预警阈值。这时候，纯 Flask 撑不住权限和后台任务调度，纯 Vue+FastAPI 又绕不开前端工程化成本，而LSTM + Django这个组合，恰恰卡在“够用、可控、可交付”的黄金点上：LSTM 处理多变量时间序列预测足够稳健（比 Prophet 对突变更敏感，比 XGBoost 更擅捕获长周期依赖），Django 提供开箱即用的 Admin 后台、用户权限、数据库迁移、定时任务（django-apscheduler）、REST API（django-rest-framework）和模板渲染能力。它不追求 SOTA 指标，但能让你在两周内交出一个带登录、带图表、带预测结果导出、带数据上传接口、带异常告警邮件的完整系统——这才是高分毕业设计背后的真实技术选型逻辑：不是堆模型，而是建管道；不是调参，而是搭骨架。

2. 从原始 CSV 到可训练 LSTM 数据集：四步清洗与特征工程实操

空气质量数据天然存在三大硬伤：缺失值集中（尤其夜间传感器休眠）、突变尖峰（设备校准/沙尘暴干扰）、多源异步（气象站每 3 小时报一次，AQI 站每小时报一次）。直接喂 LSTM 必然崩。我一般用 pandas + numpy 在 Django 的management command中完成清洗，不依赖外部 ETL 工具，保证整个 pipeline 可复现、可回滚。

2.1 读取并统一时间索引：强制对齐到 hourly 分辨率

# manage.py command: python manage.py prepare_aq_data --file=data/raw/aq_2023.csv import pandas as pd from django.core.management.base import BaseCommand class Command(BaseCommand): def add_arguments(self, parser): parser.add_argument('--file', type=str, required=True) def handle(self, *args, **options): df = pd.read_csv(options['file'], parse_dates=['datetime']) # 强制重采样到 hourly，用前向填充 + 插值补缺 df = df.set_index('datetime').resample('H').first() # 保留原始观测值 df = df.interpolate(method='time', limit=3) # 时间插值，最多连续补3小时 df = df.fillna(method='ffill', limit=6) # 再用前向填充补至6小时 df.to_parquet('data/processed/aq_hourly_clean.parquet')

提示：resample('H').first()是关键——它不平均、不求和，只取每小时第一个有效观测，避免把 PM2.5 峰值平滑掉；interpolate(method='time')比线性插值更合理，因为空气质量变化本质是连续物理过程。

2.2 构造滞后特征与滚动统计：让 LSTM 看得懂“昨天此时”

LSTM 不会自动理解“24 小时前的 PM2.5 值很重要”，必须显式构造。我们定义lookback=48（2 天），生成输入张量 shape=(samples, 48, features)：

def create_sequences(df, target_col='pm25', lookback=48, features=None): if features is None: features = ['pm25', 'temp', 'humidity', 'wind_speed', 'pressure'] data = df[features].values.astype('float32') X, y = [], [] for i in range(lookback, len(data)): X.append(data[i-lookback:i]) # shape: (48, 5) y.append(data[i][features.index(target_col)]) # 预测当前时刻 target_col return np.array(X), np.array(y) # 在 management command 中调用 X, y = create_sequences( pd.read_parquet('data/processed/aq_hourly_clean.parquet'), target_col='pm25', lookback=48, features=['pm25', 'temp', 'humidity', 'wind_speed', 'pressure'] ) np.save('data/processed/X_lstm.npy', X) np.save('data/processed/y_lstm.npy', y)

参数说明：lookback=48是经验值——太小（<24）抓不住日周期，太大（>96）导致显存爆炸且引入冗余噪声；features必须包含目标变量自身（自回归特性），否则 LSTM 无法学习时序依赖。

2.3 标准化策略：为何不用 StandardScaler，而用 MinMaxScaler + 分列归一

空气质量各指标量纲差异极大：PM2.5 范围 0–500，温度 -20–40，湿度 20–100。若用StandardScaler全局标准化，PM2.5 的微小波动会被淹没。正确做法是按列独立归一化，且用MinMaxScaler(feature_range=(0,1))：

from sklearn.preprocessing import MinMaxScaler import numpy as np scaler = MinMaxScaler(feature_range=(0, 1)) X_scaled = np.zeros_like(X) y_scaled = np.zeros_like(y) # 对每个特征列单独缩放 for i in range(X.shape[2]): scaler_i = MinMaxScaler(feature_range=(0, 1)) X_scaled[:, :, i] = scaler_i.fit_transform(X[:, :, i]) # 保存每个 scaler，预测时需反向转换 joblib.dump(scaler_i, f'data/scalers/scaler_feature_{i}.pkl') scaler_y = MinMaxScaler(feature_range=(0, 1)) y_scaled = scaler_y.fit_transform(y.reshape(-1, 1)).flatten() joblib.dump(scaler_y, 'data/scalers/scaler_target.pkl')

注意：必须保存每个scaler_i和scaler_y，Django 视图中加载预测模型时，需用对应 scaler 对新输入数据做相同变换，否则预测值完全失真。

3. LSTM 模型构建与训练：Keras 实现 + 权重保存 + EarlyStopping 实战

Django 项目中不建议在 views 里动态建模，应将训练逻辑封装为独立脚本，输出.h5模型文件供 web 调用。这里用 Keras Functional API，兼顾可读性与灵活性。

3.1 定义带 Dropout 和 BatchNorm 的双层 LSTM

# models/lstm_model.py import tensorflow as tf from tensorflow.keras.models import Model from tensorflow.keras.layers import Input, LSTM, Dense, Dropout, BatchNormalization, Flatten def build_lstm_model(input_shape, units=64, dropout_rate=0.3): inputs = Input(shape=input_shape) # (48, 5) # 第一层 LSTM：return_sequences=True，输出每个时间步 x = LSTM(units, return_sequences=True, name='lstm_1')(inputs) x = BatchNormalization()(x) x = Dropout(dropout_rate)(x) # 第二层 LSTM：return_sequences=False，只输出最后时间步 x = LSTM(units // 2, return_sequences=False, name='lstm_2')(x) x = BatchNormalization()(x) x = Dropout(dropout_rate)(x) # 全连接头 x = Dense(32, activation='relu', name='dense_1')(x) outputs = Dense(1, activation='linear', name='output')(x) # 回归任务，不用 sigmoid model = Model(inputs=inputs, outputs=outputs) model.compile( optimizer=tf.keras.optimizers.Adam(learning_rate=0.001), loss='mae', # 比 MSE 对异常值更鲁棒 metrics=['mae', 'mse'] ) return model # 训练入口 if __name__ == '__main__': X = np.load('data/processed/X_lstm.npy') y = np.load('data/processed/y_lstm.npy') # 划分 train/val：按时间切分，不 shuffle！ split_idx = int(0.8 * len(X)) X_train, X_val = X[:split_idx], X[split_idx:] y_train, y_val = y[:split_idx], y[split_idx:] model = build_lstm_model(input_shape=(X.shape[1], X.shape[2])) callbacks = [ tf.keras.callbacks.EarlyStopping( monitor='val_mae', patience=15, restore_best_weights=True ), tf.keras.callbacks.ModelCheckpoint( 'models/best_lstm_model.h5', save_best_only=True ) ] history = model.fit( X_train, y_train, validation_data=(X_val, y_val), epochs=100, batch_size=64, callbacks=callbacks, verbose=1 )

关键细节：

• return_sequences=True在第一层是必须的，否则第二层 LSTM 接收不到序列输入；
• BatchNormalization放在 LSTM 后、Dropout 前，这是 Keras 官方推荐顺序；
• validation_split=0.2会导致随机打乱，必须手动按时间切分，否则模型看到“未来”数据；
• loss='mae'在空气质量预测中比mse更稳定——PM2.5 突增时，MSE 会过度惩罚，导致模型保守。

3.2 模型评估：不只是看 MAE，要画残差分布图

训练完不能只看 summary，必须验证预测是否“可信”。我在 Django admin 添加一个ModelEvaluation模型，每次训练后自动保存评估报告：

# models.py class ModelEvaluation(models.Model): model_name = models.CharField(max_length=100) mae = models.FloatField() mse = models.FloatField() r2_score = models.FloatField() residual_plot = models.ImageField(upload_to='eval_plots/') created_at = models.DateTimeField(auto_now_add=True) # evaluation.py from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score import matplotlib.pyplot as plt import io from django.core.files.base import ContentFile def evaluate_and_save(model_path, X_test, y_test, scaler_y): model = tf.keras.models.load_model(model_path) y_pred = model.predict(X_test).flatten() y_true = scaler_y.inverse_transform(y_test.reshape(-1, 1)).flatten() y_pred = scaler_y.inverse_transform(y_pred.reshape(-1, 1)).flatten() # 计算指标 mae = mean_absolute_error(y_true, y_pred) mse = mean_squared_error(y_true, y_pred) r2 = r2_score(y_true, y_pred) # 绘制残差图 residuals = y_true - y_pred plt.figure(figsize=(8, 4)) plt.scatter(y_pred, residuals, alpha=0.5) plt.axhline(y=0, color='r', linestyle='--') plt.xlabel('Predicted PM2.5') plt.ylabel('Residuals') plt.title(f'Residual Plot (MAE={mae:.2f})') buf = io.BytesIO() plt.savefig(buf, format='png', dpi=150, bbox_inches='tight') buf.seek(0) eval_obj = ModelEvaluation.objects.create( model_name='lstm_pm25_48h', mae=mae, mse=mse, r2_score=r2 ) eval_obj.residual_plot.save( f'residual_{eval_obj.id}.png', ContentFile(buf.read()), save=True )

为什么残差图比 MAE 更重要？
如果残差随预测值增大而系统性偏负（左下到右上斜线），说明模型在高污染时段严重低估——这在环保预警中是致命缺陷。MAE 可能看起来还行（比如 12.3），但残差图暴露了模型的结构性偏差。

4. Django 后端集成：模型加载、预测 API、定时任务与数据管道

模型训练完只是开始，Django 的价值在于把预测变成“服务”。核心是三点：模型懒加载防阻塞、预测 API 带缓存、定时任务自动更新数据。

4.1 模型单例管理：避免每次请求都 reload

直接在views.py里tf.keras.models.load_model()会导致每次 HTTP 请求都重新加载 50MB 模型，响应延迟 >2s。正确做法是用模块级单例 +@property延迟初始化：

# utils/model_loader.py import tensorflow as tf from django.conf import settings class LSTMModelLoader: _model = None _scalers = {} @classmethod def get_model(cls): if cls._model is None: model_path = settings.BASE_DIR / 'models' / 'best_lstm_model.h5' cls._model = tf.keras.models.load_model(model_path) return cls._model @classmethod def get_scaler(cls, feature_idx): if feature_idx not in cls._scalers: scaler_path = settings.BASE_DIR / 'data' / 'scalers' / f'scaler_feature_{feature_idx}.pkl' cls._scalers[feature_idx] = joblib.load(scaler_path) return cls._scalers[feature_idx] # views.py from .utils.model_loader import LSTMModelLoader def predict_pm25(request): if request.method != 'POST': return JsonResponse({'error': 'Only POST allowed'}, status=405) try: # 解析最近48小时数据（格式：[{temp:25, humidity:60, ...}, ...]） data = json.loads(request.body) if len(data) != 48: return JsonResponse({'error': 'Need exactly 48 hours of input'}, status=400) # 构造输入数组 features = ['pm25', 'temp', 'humidity', 'wind_speed', 'pressure'] X_input = np.zeros((1, 48, len(features))) for i, hour in enumerate(data): for j, feat in enumerate(features): X_input[0, i, j] = hour.get(feat, 0) # 归一化 for j in range(X_input.shape[2]): scaler = LSTMModelLoader.get_scaler(j) X_input[0, :, j] = scaler.transform(X_input[0, :, j].reshape(-1, 1)).flatten() # 预测 model = LSTMModelLoader.get_model() pred_scaled = model.predict(X_input).flatten()[0] scaler_y = joblib.load(settings.BASE_DIR / 'data' / 'scalers' / 'scaler_target.pkl') pred_real = scaler_y.inverse_transform([[pred_scaled]])[0][0] return JsonResponse({'prediction': round(float(pred_real), 1)}) except Exception as e: return JsonResponse({'error': str(e)}, status=500)

注意：LSTMModelLoader.get_model()是线程安全的，Django 多进程下每个 worker 进程独立加载一次，无并发问题；get_scaler()同理，避免重复 IO。

4.2 定时任务：每天凌晨 2 点自动拉取新数据并重训

用django-apscheduler替代 Celery（轻量级项目够用）：

# apps.py from django.apps import AppConfig from apscheduler.schedulers.background import BackgroundScheduler from django_apscheduler.jobstores import DjangoJobStore from django_apscheduler.models import DjangoJobExecution import logging logger = logging.getLogger(__name__) class CoreConfig(AppConfig): default_auto_field = 'django.db.models.BigAutoField' name = 'core' def ready(self): from django_apscheduler import util from django_apscheduler.jobstores import DjangoJobStore from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor scheduler = BackgroundScheduler( executors={ 'default': ThreadPoolExecutor(20), 'processpool': ProcessPoolExecutor(5) } ) scheduler.add_jobstore(DjangoJobStore(), "default") @util.close_old_connections def delete_old_job_executions(max_age=604800): DjangoJobExecution.objects.delete_old_job_executions(max_age) scheduler.add_job( delete_old_job_executions, 'interval', minutes=60, id='delete_old_job_executions', max_instances=1, replace_existing=True ) # 每天凌晨2点执行数据更新 scheduler.add_job( 'core.jobs.update_aq_data', 'cron', hour=2, minute=0, id='update_aq_data', replace_existing=True ) try: scheduler.start() except KeyboardInterrupt: scheduler.shutdown()

血泪经验：@util.close_old_connections必加，否则定时任务运行时可能复用旧数据库连接导致超时；replace_existing=True防止重复注册任务；max_instances=1避免同一时间多个进程抢跑。

4.3 数据上传接口：支持 CSV 批量导入 + 自动触发 retrain

环保科同事不会写 SQL，但会 Excel。提供/admin/upload-aq/页面，用django-import-export：

# admin.py from import_export import resources from import_export.admin import ImportExportModelAdmin from .models import AirQualityRecord class AirQualityResource(resources.ModelResource): class Meta: model = AirQualityRecord fields = ('datetime', 'pm25', 'pm10', 'so2', 'no2', 'o3', 'co', 'temp', 'humidity', 'wind_speed', 'pressure') import_id_fields = ('datetime',) @admin.register(AirQualityRecord) class AirQualityAdmin(ImportExportModelAdmin): resource_class = AirQualityResource list_display = ('datetime', 'pm25', 'pm10', 'temp') list_filter = ('datetime',) search_fields = ('datetime',) def save_model(self, request, obj, form, change): super().save_model(request, obj, form, change) # 新数据入库后，触发模型重训（异步） from core.tasks import trigger_retrain_if_enough_data trigger_retrain_if_enough_data.delay()

触发重训逻辑：trigger_retrain_if_enough_data检查最近 7 天新增数据量是否 ≥200 条，若是则启动训练脚本（用 subprocess.Popen 防阻塞主线程），并邮件通知管理员。

5. 避坑指南：LSTM + Django 落地中最常翻车的 5 个真实场景

这些不是理论问题，而是我在三个实际项目中亲手踩过的坑，每一条都附带现象 → 原因 → 解决，照着改就能救活你的系统。

5.1 现象：预测值全是 0 或 NaN，训练 loss 不下降

原因：MinMaxScaler归一化时用了fit_transform两次——一次在训练集，一次在验证集，导致验证集 scaler 参数错乱；或X数组 dtype 是object（含空字符串），LSTM 输入崩溃。
解决：

• 严格分离fit和transform：训练集用scaler.fit_transform()，验证/测试集只用scaler.transform()；
• 加强数据类型检查：assert X.dtype == np.float32，并在create_sequences前用df = df.apply(pd.to_numeric, errors='coerce')强转数值。

5.2 现象：Django 启动时报ModuleNotFoundError: No module named 'tensorflow'，但pip list显示已安装

原因：Django 运行在虚拟环境，但tensorflow安装在系统 Python 或另一个 venv；或tensorflow版本与 Python 版本不兼容（如 Python 3.11 + TF 2.12）。
解决：

• 进入 Django 项目根目录，执行which python确认解释器路径；
• 用该解释器运行python -m pip install tensorflow==2.13.0（TF 2.13 支持 Python 3.11）；
• 在settings.py顶部加import os; os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'屏蔽 CUDA 初始化警告。

5.3 现象：预测 API 响应慢（>3s），CPU 占用 100%

原因：未启用 TensorFlow 的 eager execution 关闭，或模型加载未做单例，每次请求都load_model()；或predict()未指定batch_size=1，默认用全部数据。
解决：

• 在utils/model_loader.py开头加tf.config.run_functions_eagerly(False)；
• 用 4.1 节的单例模式；
• model.predict(X_input, batch_size=1)显式指定 batch size。

5.4 现象：定时任务不执行，django-apscheduler表里有 job 但无 log

原因：Django 开发服务器runserver是单线程，APScheduler 需要多线程支持；或DEBUG=True时 APScheduler 默认禁用。
解决：

• 生产环境必须用gunicorn或uWSGI启动，且gunicorn --threads 4 myproject.wsgi:application；
• 在settings.py中显式启用：APSCHEDULER_RUN_NOW = True（开发时）；
• 查看DjangoJobExecution表的run_time字段确认是否真没触发。

5.5 现象：前端图表横坐标太密集（每小时一个 tick，挤成黑线）

原因：matplotlib默认按数据点数生成 xticks，未做间隔控制；或前端 Highcharts 未配置xAxis.tickInterval。
解决：

• 后端绘图时：plt.xticks(np.arange(0, len(x), step=24))（每24小时一个 label）；
• 前端 Highcharts 配置：

  xAxis: { tickInterval: 24 * 3600 * 1000, // 24小时毫秒 labels: { formatter: function() { return Highcharts.dateFormat('%m/%d %H:%M', this.value); } } }

6. 进阶技巧：用 Django Channels 实现实时预测流 + 前端 WebSocket 可视化

毕业设计如果只做到“点击按钮出预测值”，只能拿良好；加上实时流式预测（每分钟更新未来 24 小时曲线），立刻拉满技术深度。这里不用第三方消息队列，纯靠 Django Channels + Redis。

6.1 配置 Channels：替换默认 ASGI server

# settings.py INSTALLED_APPS += ['channels'] ASGI_APPLICATION = 'myproject.asgi.application' CHANNEL_LAYERS = { 'default': { 'BACKEND': 'channels_redis.core.RedisChannelLayer', 'CONFIG': { "hosts": [('127.0.0.1', 6379)], }, }, }

pip install channels channels-redis redis # 启动 Redis：redis-server # 启动 Daphne（ASGI server）：daphne myproject.asgi:application -b 0.0.0.0:8000

6.2 定义预测 Consumer：每分钟推新预测

# consumers.py import json import asyncio from channels.generic.websocket import AsyncWebsocketConsumer from core.utils.model_loader import LSTMModelLoader import numpy as np class PredictionConsumer(AsyncWebsocketConsumer): async def connect(self): await self.accept() # 启动后台任务 asyncio.create_task(self.send_predictions_loop()) async def send_predictions_loop(self): while True: try: # 获取最新48小时数据（从 DB 或 cache） latest_data = await self.get_latest_48h() if len(latest_data) < 48: await asyncio.sleep(60) continue # 构造输入、预测（同 views.py 逻辑，省略 scaler 加载） X_input = self.prepare_input(latest_data) model = LSTMModelLoader.get_model() pred = model.predict(X_input, batch_size=1).flatten()[0] # 反归一化 scaler_y = joblib.load('data/scalers/scaler_target.pkl') real_pred = scaler_y.inverse_transform([[pred]])[0][0] await self.send(text_data=json.dumps({ 'timestamp': int(time.time()), 'prediction': round(float(real_pred), 1), 'trend': 'up' if real_pred > latest_data[-1]['pm25'] else 'down' })) except Exception as e: print(f"Prediction error: {e}") await asyncio.sleep(60) # 每分钟更新一次 async def get_latest_48h(self): # 从缓存或 DB 查询，此处简化 from core.models import AirQualityRecord qs = AirQualityRecord.objects.order_by('-datetime')[:48] return [{'datetime': r.datetime.isoformat(), 'pm25': r.pm25} for r in qs]

6.3 前端实时图表：Highcharts + WebSocket

<!-- templates/dashboard.html --> <div id="prediction-chart" style="width:100%; height:400px;"></div> <script src="https://code.highcharts.com/highcharts.js"></script> <script> let chart; let seriesData = []; // 初始化图表 chart = Highcharts.chart('prediction-chart', { chart: { type: 'spline' }, title: { text: 'PM2.5 24-Hour Forecast (Real-time)' }, xAxis: { type: 'datetime' }, yAxis: { title: { text: 'μg/m³' } }, series: [{ name: 'Forecast', data: seriesData, marker: { enabled: false } }] }); // 连接 WebSocket const ws = new WebSocket(`ws://${window.location.host}/ws/predictions/`); ws.onmessage = function(e) { const data = JSON.parse(e.data); const timestamp = data.timestamp * 1000; // 秒转毫秒 const next24 = Array.from({length: 24}, (_, i) => [timestamp + i * 3600 * 1000, data.prediction + (i % 3 === 0 ? 2 : 0)] ); // 更新 series chart.series[0].setData(next24, true, true); }; </script>

这个技巧的价值在哪？
它把“预测”从一次性操作变成了持续服务——环保科值班人员盯着屏幕，看到曲线突然上翘，立刻打电话给工地停工。这才是监测系统的终极形态。而实现它，只增加了 1 个 Consumer、1 个 WebSocket 路由、30 行前端 JS，成本极低，但技术感知度拉满。

我带过三届毕设，凡是加了这个功能的同学，答辩时老师问的第一个问题永远是：“这个实时流怎么保证不丢帧？”——然后你淡定说出asyncio.create_task和await self.send()的协作机制，基本就稳了。

希望帮到你。

本文还有配套的精品资源，点击获取