企业官网建设流程全解析

Python智慧交通客流量分析预测实战：从数据采集到可视化大屏

每到毕业季，计算机专业的学生最头疼的就是毕业设计选题。既要体现技术含量，又要具备实际应用价值，还要能展示自己的编程能力。如果你正在为"智慧交通"方向的毕业设计发愁，那么客流量分析预测这个课题绝对值得考虑。

这个项目之所以被称为"吊炸天"，不是因为它用了多高深的技术，而是它完美结合了Python数据分析、机器学习算法和可视化技术，解决的是城市交通管理中的真实痛点。更重要的是，这个项目的技术栈正是当前企业招聘的热门需求：Python数据处理、机器学习建模、FlaskWeb开发、数据可视化。

本文将带你完整实现一个智慧交通客流量分析预测系统，从数据爬取、预处理、模型训练到可视化展示，每个环节都有可运行的代码示例。无论你是毕业设计急需参考，还是想学习数据分析全流程，这篇文章都能给你实实在在的帮助。

1. 项目整体架构设计

智慧交通客流量分析预测系统的核心价值在于将抽象的交通数据转化为直观的决策支持信息。整个系统可以分为四个关键模块：

数据采集层：负责从多种数据源获取原始交通数据，包括实时客流数据、历史交通流量、天气信息、节假日数据等。这些数据通常以CSV、JSON或数据库形式存储。

数据处理层：对原始数据进行清洗、转换和特征工程。这一层要处理缺失值、异常值，进行数据标准化，并提取有预测价值的特征，如时间特征（小时、星期、是否节假日）、空间特征（区域、路段）等。

模型预测层：使用机器学习算法建立预测模型。线性回归作为基线模型，后续可以扩展至更复杂的算法如随机森林、XGBoost、LSTM等。这一层负责训练模型并生成预测结果。

可视化展示层：通过Web界面展示数据分析结果和预测趋势，使用ECharts、Pyecharts等可视化库制作交互式图表和大屏展示。

# 项目目录结构 traffic_analysis/ ├── data/ # 数据目录 │ ├── raw/ # 原始数据 │ ├── processed/ # 处理后的数据 │ └── models/ # 训练好的模型 ├── src/ # 源代码 │ ├── data_processing.py # 数据处理 │ ├── model_training.py # 模型训练 │ ├── prediction.py # 预测模块 │ └── visualization.py # 可视化 ├── app.py # Flask主应用 ├── requirements.txt # 依赖包 └── config.py # 配置文件

这种分层架构的优势在于模块化设计，每个部分可以独立开发和测试，也便于后续的功能扩展和维护。

2. 环境准备与依赖安装

开始之前，我们需要配置完整的Python开发环境。推荐使用Python 3.8+版本，这个版本在稳定性和库兼容性方面表现良好。

2.1 创建虚拟环境

虚拟环境可以隔离项目依赖，避免版本冲突问题：

# 创建虚拟环境 python -m venv traffic_env # 激活虚拟环境（Windows） traffic_env\Scripts\activate # 激活虚拟环境（Mac/Linux） source traffic_env/bin/activate

2.2 安装依赖包

创建requirements.txt文件，包含项目所需的所有依赖：

# requirements.txt flask==2.3.3 pandas==2.0.3 numpy==1.24.3 scikit-learn==1.3.0 matplotlib==3.7.2 seaborn==0.12.2 pyecharts==2.0.3 jupyter==1.0.0 requests==2.31.0 openpyxl==3.1.2

安装依赖：

pip install -r requirements.txt

2.3 验证安装

创建一个简单的验证脚本来检查关键库是否正常导入：

# check_environment.py try: import flask import pandas as pd import numpy as np from sklearn.linear_model import LinearRegression import pyecharts print("所有依赖库导入成功！") print(f"Flask版本: {flask.__version__}") print(f"Pandas版本: {pd.__version__}") except ImportError as e: print(f"导入错误: {e}")

环境配置中最常见的坑是版本冲突。如果遇到问题，可以尝试先安装基础版本，再逐步添加其他依赖。

3. 数据采集与预处理实战

高质量的数据是分析预测的基础。智慧交通项目通常需要多源数据融合，包括历史客流数据、时间特征、天气信息等。

3.1 模拟交通数据生成

在实际项目中，我们可以从公开数据源或企业内部系统获取数据。这里我们先生成模拟数据用于演示：

# data_generation.py import pandas as pd import numpy as np from datetime import datetime, timedelta def generate_traffic_data(num_days=365): """生成模拟交通客流数据""" dates = pd.date_range(start='2023-01-01', periods=num_days, freq='D') data = [] for i, date in enumerate(dates): # 基础客流（受星期影响） base_traffic = 1000 weekday_effect = 200 if date.weekday() < 5 else 500 # 周末客流增加 # 季节性效应（模拟节假日） seasonal_effect = 300 * np.sin(2 * np.pi * i / 365) # 随机噪声 noise = np.random.normal(0, 50) # 特殊事件（节假日效应） holiday_effect = 0 if date.month == 10 and date.day in [1, 2, 3]: # 国庆节 holiday_effect = 800 elif date.month == 1 and date.day == 1: # 元旦 holiday_effect = 600 # 计算总客流 total_traffic = base_traffic + weekday_effect + seasonal_effect + holiday_effect + noise data.append({ 'date': date, 'traffic_volume': max(500, total_traffic), # 确保非负 'weekday': date.weekday(), 'is_weekend': 1 if date.weekday() >= 5 else 0, 'month': date.month, 'day_of_year': i, 'is_holiday': 1 if holiday_effect > 0 else 0 }) return pd.DataFrame(data) # 生成并保存数据 df = generate_traffic_data() df.to_csv('data/raw/traffic_data.csv', index=False) print("模拟数据生成完成，共生成{}条记录".format(len(df)))

3.2 数据清洗与特征工程

原始数据往往存在缺失值、异常值等问题，需要进行预处理：

# data_processing.py import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler class TrafficDataProcessor: def __init__(self): self.scaler = StandardScaler() def load_data(self, file_path): """加载数据""" self.df = pd.read_csv(file_path) self.df['date'] = pd.to_datetime(self.df['date']) return self.df def handle_missing_values(self): """处理缺失值""" # 检查缺失值 missing_info = self.df.isnull().sum() print("缺失值统计:") print(missing_info) # 对数值列用中位数填充 numeric_cols = self.df.select_dtypes(include=[np.number]).columns self.df[numeric_cols] = self.df[numeric_cols].fillna(self.df[numeric_cols].median()) return self.df def detect_outliers(self, column='traffic_volume'): """检测异常值""" Q1 = self.df[column].quantile(0.25) Q3 = self.df[column].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR outliers = self.df[(self.df[column] < lower_bound) | (self.df[column] > upper_bound)] print(f"检测到{len(outliers)}个异常值") return outliers def create_features(self): """创建特征工程""" # 时间特征 self.df['year'] = self.df['date'].dt.year 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['traffic_lag1'] = self.df['traffic_volume'].shift(1) self.df['traffic_lag7'] = self.df['traffic_volume'].shift(7) # 移动平均特征 self.df['traffic_ma7'] = self.df['traffic_volume'].rolling(window=7).mean() # 删除因创建滞后特征产生的缺失值 self.df = self.df.dropna() return self.df def prepare_model_data(self): """准备模型训练数据""" # 选择特征列 feature_cols = ['weekday', 'is_weekend', 'month_sin', 'month_cos', 'traffic_lag1', 'traffic_lag7', 'traffic_ma7', 'is_holiday'] X = self.df[feature_cols] y = self.df['traffic_volume'] return X, y # 使用示例 if __name__ == "__main__": processor = TrafficDataProcessor() df = processor.load_data('data/raw/traffic_data.csv') df = processor.handle_missing_values() outliers = processor.detect_outliers() df = processor.create_features() X, y = processor.prepare_model_data() print("数据预处理完成") print(f"特征维度: {X.shape}") print(f"目标变量维度: {y.shape}")

特征工程是机器学习项目中最重要的环节之一。好的特征能够显著提升模型性能，而糟糕的特征即使使用最先进的算法也难以取得好结果。

4. 线性回归模型构建与训练

线性回归虽然简单，但在时间序列预测中往往能提供不错的基线效果，而且模型可解释性强。

4.1 基础线性回归实现

# model_training.py import pandas as pd import numpy as np from sklearn.linear_model import LinearRegression from sklearn.model_selection import train_test_split from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score import matplotlib.pyplot as plt import seaborn as sns import joblib class TrafficPredictor: def __init__(self): self.model = LinearRegression() self.feature_importance = None def train_test_split(self, X, y, test_size=0.2): """划分训练集和测试集""" # 对于时间序列数据，按时间顺序划分 split_point = int(len(X) * (1 - test_size)) X_train, X_test = X.iloc[:split_point], X.iloc[split_point:] y_train, y_test = y.iloc[:split_point], y.iloc[split_point:] return X_train, X_test, y_train, y_test def train_model(self, X_train, y_train): """训练模型""" self.model.fit(X_train, y_train) # 计算特征重要性（系数的绝对值） self.feature_importance = pd.DataFrame({ 'feature': X_train.columns, 'importance': abs(self.model.coef_) }).sort_values('importance', ascending=False) return self.model def evaluate_model(self, X_test, y_test): """评估模型性能""" y_pred = self.model.predict(X_test) metrics = { 'MAE': mean_absolute_error(y_test, y_pred), 'MSE': mean_squared_error(y_test, y_pred), 'RMSE': np.sqrt(mean_squared_error(y_test, y_pred)), 'R2': r2_score(y_test, y_pred) } # 创建评估结果DataFrame results_df = pd.DataFrame({ 'Actual': y_test.values, 'Predicted': y_pred, 'Date': X_test.index if hasattr(X_test, 'index') else range(len(y_test)) }) return metrics, results_df def plot_predictions(self, results_df): """绘制预测结果对比图""" plt.figure(figsize=(12, 6)) plt.subplot(1, 2, 1) plt.plot(results_df['Actual'].values, label='实际值', alpha=0.7) plt.plot(results_df['Predicted'].values, label='预测值', alpha=0.7) plt.title('客流量预测对比') plt.xlabel('样本索引') plt.ylabel('客流量') plt.legend() plt.subplot(1, 2, 2) plt.scatter(results_df['Actual'], results_df['Predicted'], alpha=0.6) plt.plot([results_df['Actual'].min(), results_df['Actual'].max()], [results_df['Actual'].min(), results_df['Actual'].max()], 'r--') plt.title('预测值 vs 实际值') plt.xlabel('实际客流量') plt.ylabel('预测客流量') plt.tight_layout() plt.show() def save_model(self, file_path): """保存训练好的模型""" joblib.dump(self.model, file_path) print(f"模型已保存到 {file_path}") def load_model(self, file_path): """加载已训练的模型""" self.model = joblib.load(file_path) print(f"模型已从 {file_path} 加载") # 完整的训练流程示例 def complete_training_example(): # 数据准备 from data_processing import TrafficDataProcessor processor = TrafficDataProcessor() df = processor.load_data('data/raw/traffic_data.csv') df = processor.handle_missing_values() df = processor.create_features() X, y = processor.prepare_model_data() # 模型训练 predictor = TrafficPredictor() X_train, X_test, y_train, y_test = predictor.train_test_split(X, y) print(f"训练集大小: {len(X_train)}") print(f"测试集大小: {len(X_test)}") # 训练模型 predictor.train_model(X_train, y_train) # 评估模型 metrics, results_df = predictor.evaluate_model(X_test, y_test) print("\n模型评估结果:") for metric, value in metrics.items(): print(f"{metric}: {value:.4f}") # 显示特征重要性 print("\n特征重要性排序:") print(predictor.feature_importance) # 绘制结果 predictor.plot_predictions(results_df) # 保存模型 predictor.save_model('data/models/linear_regression_model.pkl') return predictor, metrics if __name__ == "__main__": predictor, metrics = complete_training_example()

4.2 模型优化与交叉验证

基础线性回归可能无法捕捉复杂的时间模式，我们可以通过交叉验证和正则化来优化模型：

# model_optimization.py from sklearn.linear_model import Ridge, Lasso from sklearn.model_selection import TimeSeriesSplit, cross_val_score from sklearn.preprocessing import StandardScaler class AdvancedTrafficPredictor(TrafficPredictor): def __init__(self, model_type='ridge'): if model_type == 'ridge': self.model = Ridge(alpha=1.0) elif model_type == 'lasso': self.model = Lasso(alpha=0.1) else: self.model = LinearRegression() self.scaler = StandardScaler() self.model_type = model_type def time_series_cross_validation(self, X, y, n_splits=5): """时间序列交叉验证""" tscv = TimeSeriesSplit(n_splits=n_splits) scores = [] for train_index, test_index in tscv.split(X): X_train, X_test = X.iloc[train_index], X.iloc[test_index] y_train, y_test = y.iloc[train_index], y.iloc[test_index] # 标准化特征 X_train_scaled = self.scaler.fit_transform(X_train) X_test_scaled = self.scaler.transform(X_test) # 训练和评估 self.model.fit(X_train_scaled, y_train) score = self.model.score(X_test_scaled, y_test) scores.append(score) return np.mean(scores), np.std(scores) def hyperparameter_tuning(self, X, y): """超参数调优""" alphas = [0.001, 0.01, 0.1, 1, 10, 100] best_score = -np.inf best_alpha = alphas[0] for alpha in alphas: if self.model_type == 'ridge': model = Ridge(alpha=alpha) elif self.model_type == 'lasso': model = Lasso(alpha=alpha) else: continue # 使用交叉验证评估 tscv = TimeSeriesSplit(n_splits=5) cv_scores = cross_val_score(model, self.scaler.fit_transform(X), y, cv=tscv, scoring='r2') mean_score = cv_scores.mean() if mean_score > best_score: best_score = mean_score best_alpha = alpha print(f"最佳alpha值: {best_alpha}, 最佳R2分数: {best_score:.4f}") return best_alpha # 使用优化后的模型 def optimized_training(): from data_processing import TrafficDataProcessor processor = TrafficDataProcessor() df = processor.load_data('data/raw/traffic_data.csv') df = processor.handle_missing_values() df = processor.create_features() X, y = processor.prepare_model_data() # 使用岭回归 advanced_predictor = AdvancedTrafficPredictor(model_type='ridge') # 超参数调优 best_alpha = advanced_predictor.hyperparameter_tuning(X, y) advanced_predictor.model.set_params(alpha=best_alpha) # 交叉验证 mean_score, std_score = advanced_predictor.time_series_cross_validation(X, y) print(f"交叉验证平均R2: {mean_score:.4f} (±{std_score:.4f})") # 最终训练 X_train, X_test, y_train, y_test = advanced_predictor.train_test_split(X, y) X_train_scaled = advanced_predictor.scaler.fit_transform(X_train) X_test_scaled = advanced_predictor.scaler.transform(X_test) advanced_predictor.model.fit(X_train_scaled, y_train) # 评估 y_pred = advanced_predictor.model.predict(X_test_scaled) r2 = r2_score(y_test, y_pred) print(f"测试集R2分数: {r2:.4f}") return advanced_predictor

5. Flask Web应用开发

有了预测模型，我们需要一个Web界面来展示结果。Flask是一个轻量级的Python Web框架，非常适合这类数据展示应用。

5.1 基础Flask应用结构

# app.py from flask import Flask, render_template, request, jsonify import pandas as pd import numpy as np import joblib from datetime import datetime, timedelta import json app = Flask(__name__) # 加载训练好的模型和数据处理器 try: model = joblib.load('data/models/linear_regression_model.pkl') print("模型加载成功") except: print("模型文件不存在，请先训练模型") model = None class TrafficAnalysisSystem: def __init__(self): self.model = model self.df = None def load_sample_data(self): """加载示例数据用于演示""" # 这里可以替换为实际的数据加载逻辑 dates = pd.date_range(start='2023-01-01', end='2023-12-31', freq='D') traffic_data = np.random.randint(800, 2000, size=len(dates)) self.df = pd.DataFrame({ 'date': dates, 'traffic_volume': traffic_data }) return self.df def predict_future(self, days=30): """预测未来客流""" if self.model is None: return None # 这里需要根据实际模型特征进行预测 # 简化示例：返回随机预测值 future_dates = pd.date_range(start=datetime.now(), periods=days, freq='D') predictions = np.random.randint(1000, 1800, size=days) return pd.DataFrame({ 'date': future_dates, 'predicted_traffic': predictions }) # 初始化系统 traffic_system = TrafficAnalysisSystem() @app.route('/') def index(): """主页""" return render_template('index.html') @app.route('/api/traffic_data') def get_traffic_data(): """获取交通数据API""" df = traffic_system.load_sample_data() # 转换为前端需要的格式 chart_data = { 'dates': df['date'].dt.strftime('%Y-%m-%d').tolist(), 'traffic': df['traffic_volume'].tolist() } return jsonify(chart_data) @app.route('/api/predict', methods=['POST']) def predict_traffic(): """预测客流API""" data = request.json days = data.get('days', 30) predictions = traffic_system.predict_future(days) if predictions is not None: result = { 'dates': predictions['date'].dt.strftime('%Y-%m-%d').tolist(), 'predictions': predictions['predicted_traffic'].tolist() } return jsonify(result) else: return jsonify({'error': '模型未加载'}), 500 @app.route('/dashboard') def dashboard(): """数据大屏页面""" return render_template('dashboard.html') if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000)

5.2 前端模板开发

创建 templates/index.html 文件：

<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>智慧交通客流量分析系统</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> <style> body { font-family: Arial, sans-serif; margin: 0; padding: 20px; background-color: #f5f5f5; } .container { max-width: 1200px; margin: 0 auto; background: white; padding: 20px; border-radius: 8px; } .chart-container { height: 400px; margin: 20px 0; } .header { text-align: center; margin-bottom: 30px; } .controls { margin: 20px 0; } button { padding: 10px 20px; background: #007bff; color: white; border: none; border-radius: 4px; cursor: pointer; } button:hover { background: #0056b3; } </style> </head> <body> <div class="container"> <div class="header"> <h1>智慧交通客流量分析预测系统</h1> <p>基于Python和机器学习的交通大数据分析平台</p> </div> <div class="controls"> <label>预测天数: </label> <input type="number" id="daysInput" value="30" min="7" max="90"> <button onclick="loadPredictions()">生成预测</button> </div> <div id="trafficChart" class="chart-container"></div> <div id="predictionChart" class="chart-container"></div> </div> <script> // 初始化ECharts实例 const trafficChart = echarts.init(document.getElementById('trafficChart')); const predictionChart = echarts.init(document.getElementById('predictionChart')); // 加载历史数据 fetch('/api/traffic_data') .then(response => response.json()) .then(data => { const option = { title: { text: '历史客流量趋势' }, tooltip: { trigger: 'axis' }, xAxis: { type: 'category', data: data.dates }, yAxis: { type: 'value' }, series: [{ data: data.traffic, type: 'line', smooth: true, areaStyle: {} }] }; trafficChart.setOption(option); }); function loadPredictions() { const days = document.getElementById('daysInput').value; fetch('/api/predict', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ days: parseInt(days) }) }) .then(response => response.json()) .then(data => { const option = { title: { text: `未来${days}天客流量预测` }, tooltip: { trigger: 'axis' }, xAxis: { type: 'category', data: data.dates }, yAxis: { type: 'value' }, series: [{ data: data.predictions, type: 'line', smooth: true, lineStyle: { color: '#ff6b6b' } }] }; predictionChart.setOption(option); }); } // 初始化预测图表 loadPredictions(); // 响应窗口大小变化 window.addEventListener('resize', function() { trafficChart.resize(); predictionChart.resize(); }); </script> </body> </html>

6. 高级可视化与Pyecharts应用

Pyecharts是基于ECharts的Python可视化库，可以创建交互性更强的图表。

6.1 使用Pyecharts创建高级图表

# visualization.py from pyecharts import options as opts from pyecharts.charts import Line, Bar, Pie, Grid, Page import pandas as pd class TrafficVisualizer: def __init__(self, df): self.df = df def create_traffic_trend_chart(self): """创建客流趋势图""" dates = self.df['date'].dt.strftime('%Y-%m-%d').tolist() traffic = self.df['traffic_volume'].tolist() line = ( Line() .add_xaxis(dates) .add_yaxis("客流量", traffic, is_smooth=True, linestyle_opts=opts.LineStyleOpts(width=3), itemstyle_opts=opts.ItemStyleOpts(color="#5793f3")) .set_global_opts( title_opts=opts.TitleOpts(title="客流量趋势分析"), tooltip_opts=opts.TooltipOpts(trigger="axis"), datazoom_opts=opts.DataZoomOpts(), yaxis_opts=opts.AxisOpts( name="客流量", type_="value", axistick_opts=opts.AxisTickOpts(is_show=True), splitline_opts=opts.SplitLineOpts(is_show=True), ), xaxis_opts=opts.AxisOpts( name="日期", type_="category", boundary_gap=False, ), ) ) return line def create_weekly_analysis_chart(self): """创建周分析图表""" self.df['weekday_name'] = self.df['date'].dt.day_name() weekly_avg = self.df.groupby('weekday_name')['traffic_volume'].mean() # 按星期顺序排序 weekday_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'] weekly_avg = weekly_avg.reindex(weekday_order) bar = ( Bar() .add_xaxis(weekly_avg.index.tolist()) .add_yaxis("平均客流量", weekly_avg.values.tolist(), itemstyle_opts=opts.ItemStyleOpts(color="#91cc75")) .set_global_opts( title_opts=opts.TitleOpts(title="每周客流量分析"), xaxis_opts=opts.AxisOpts(name="星期"), yaxis_opts=opts.AxisOpts(name="平均客流量"), ) ) return bar def create_seasonal_analysis_chart(self): """创建季节性分析图表""" monthly_avg = self.df.groupby(self.df['date'].dt.month)['traffic_volume'].mean() line = ( Line() .add_xaxis([f"{int(month)}月" for month in monthly_avg.index]) .add_yaxis("月平均客流量", monthly_avg.values.tolist(), is_smooth=True, linestyle_opts=opts.LineStyleOpts(width=3), itemstyle_opts=opts.ItemStyleOpts(color="#fac858")) .set_global_opts( title_opts=opts.TitleOpts(title="季节性客流分析"), xaxis_opts=opts.AxisOpts(name="月份"), yaxis_opts=opts.AxisOpts(name="平均客流量"), ) ) return line def create_dashboard(self): """创建综合仪表板""" page = Page(layout=Page.DraggablePageLayout) page.add( self.create_traffic_trend_chart(), self.create_weekly_analysis_chart(), self.create_seasonal_analysis_chart() ) return page # 使用示例 def create_visualization_demo(): # 生成示例数据 dates = pd.date_range(start='2023-01-01', end='2023-12-31', freq='D') traffic_data = 1000 + 500 * np.sin(2 * np.pi * np.arange(len(dates)) / 365) + np.random.normal(0, 50, len(dates)) df = pd.DataFrame({ 'date': dates, 'traffic_volume': traffic_data }) visualizer = TrafficVisualizer(df) dashboard = visualizer.create_dashboard() dashboard.render("traffic_dashboard.html") print("可视化仪表板已生成: traffic_dashboard.html") if __name__ == "__main__": create_visualization_demo()

6.2 实时数据可视化大屏

对于智慧交通系统，实时数据监控大屏是重要组成部分：

# realtime_dashboard.py from pyecharts.charts import Line, Gauge, EffectScatter, Grid from pyecharts import options as opts import random import time from flask import Flask, render_template_string, jsonify import threading class RealTimeDashboard: def __init__(self): self.current_traffic = 1000 self.traffic_history = [] self.update_interval = 2 # 更新间隔（秒） def generate_realtime_data(self): """生成实时数据""" # 模拟实时客流变化 change = random.randint(-50, 50) self.current_traffic = max(500, self.current_traffic + change) self.traffic_history.append(self.current_traffic) # 保持历史数据长度 if len(self.traffic_history) > 60: # 保留最近60个数据点 self.traffic_history.pop(0) return { 'current': self.current_traffic, 'history': self.traffic_history[-30:], # 最近30个点用于图表 'timestamp': time.time() } def create_realtime_chart(self, data): """创建实时图表""" times = [f"{i}s前" for i in range(len(data['history']), 0, -1)] line = ( Line() .add_xaxis(times) .add_yaxis("实时客流量", data['history'], is_smooth=True, linestyle_opts=opts.LineStyleOpts(width=3), itemstyle_opts=opts.ItemStyleOpts(color="#c23531")) .set_global_opts( title_opts=opts.TitleOpts(title="实时客流量监控", pos_left="center"), tooltip_opts=opts.TooltipOpts(trigger="axis"), yaxis_opts=opts.AxisOpts( type_="value", min_=500, max_=2000 ), ) ) # 仪表盘显示当前值 gauge = ( Gauge() .add( "当前客流", [("", data['current'])], axisline_opts=opts.AxisLineOpts( linestyle_opts=opts.LineStyleOpts( color=[[0.3, "#67e0e3"], [0.7, "#37a2da"], [1, "#fd666d"]], width=30 ) ), detail_label_opts=opts.LabelOpts(formatter="{value}人"), ) .set_global_opts( title_opts=opts.TitleOpts(title="实时客流指标"), legend_opts=opts.LegendOpts(is_show=False), ) ) return line, gauge # Flask实时数据接口 app = Flask(__name__) dashboard = RealTimeDashboard() @app.route('/') def index(): """实时监控页面""" html_template = """ <!DOCTYPE html> <html> <head> <title>实时交通监控</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> <style> body { margin: 0; padding: 20px; background: #1a1a1a; color: white; } .chart { height: 400px; margin: 20px 0; } .grid-container { display: grid; grid-template-columns: 2fr 1fr; gap: 20px; } </style> </head> <body> <h1>智慧交通实时监控大屏</h1> <div class="grid-container"> <div id="lineChart" class="chart"></div> <div id="gaugeChart" class="chart"></div> </div> <script> let lineChart = echarts.init(document.getElementById('lineChart