1、项目介绍

技术栈：Python语言、Flask框架、Vue前端框架、MySQL数据库、Echarts可视化、requests爬虫技术、Arima算法、LSTM算法。

功能模块：

• 首页仪表盘：展示核心统计数据、客流量柱状图、城市健康状态占比饼图、客流前十城市趋势折线图。
• 交通数据分析：包含中国主要城市交通健康榜和城市详情数据分析，支持多指标筛选与图表联动。
• 地铁数据分析：支持城市切换，展示区域客流数据、历史客流面积图及时段客流统计表格。
• 模型对比：支持城市选择，对比ARIMA与LSTM模型的评估指标及预测拟合效果。
• 地铁数据预测：支持城市选择，基于ARIMA与LSTM输出未来一周客流量预测折线图。
• 交通数据预测：支持城市选择，预测未来一周城市拥堵指数，展示历史与预测趋势折线图。
• 登录页面：提供账号密码验证入口，保障系统访问安全。

项目介绍：本系统针对城市交通拥堵问题，设计并实现了一套交通数据分析与预测平台。系统基于Python、Flask、Vue和MySQL构建，通过requests爬虫采集MetroDB网站的交通健康榜及地铁运营数据，经清洗后存储于数据库。系统集成了首页仪表盘、交通数据分析、地铁数据分析、模型对比、地铁与交通数据预测等多个功能模块，利用ARIMA与LSTM算法实现短期客流量与拥堵趋势预测，并通过Echarts提供多类型图表可视化展示，支持城市选择、数据刷新等交互操作，为交通管理决策与公众出行规划提供全面的数据支撑。

2、项目界面

一、首页仪表盘，交通数据
该页面为交通大数据分析系统的地铁数据分析可视化页面，可选择对应城市，通过仪表盘、柱状图等多种图表，直观呈现地铁客流、线路站点、运营里程等多维度数据，支持查看近期客流趋势与历年客流情况，实现地铁交通数据的可视化分析。

二、首页仪表盘
该页面为交通大数据分析系统的主页仪表盘，顶部展示地铁相关核心运营数据统计，通过柱状图呈现不同星期的客流量情况，以饼图展示城市健康状态占比，用折线图呈现客流前十城市的趋势，实现交通数据的全局可视化概览与多维度分析。

三、中国主要城市交通健康榜
该页面为交通大数据分析系统的交通数据分析模块，呈现中国主要城市交通健康榜的详细数据表格，展示城市交通健康指数、拥堵指数等指标，同时通过饼图展示城市健康状态占比，实现城市交通运行状态的多维度数据可视化分析。

四、交通数据分析
该页面为交通大数据分析系统的交通数据分析模块，可选择目标城市与时间、指标维度，通过折线图展示交通拥堵指数的时段变化趋势，以表格呈现区域拥堵排名，结合热力地图直观呈现各区域拥堵分布，实现城市交通运行状态的多维度可视化分析。

五、地铁数据分析
该页面属于交通大数据分析系统的地铁数据分析模块，支持切换城市查看区域客流数据，以面积图表呈现地铁历史客流量的走势趋势，辅以分页表格展示各城市的时段客流统计，实现对交通客流的多维度可视化分析与趋势监控。

六、模型对比
该页面为交通大数据分析系统的模型对比分析模块，可选择目标城市，展示ARIMA与LSTM两种预测模型的评估指标，通过折线图分别呈现两种模型的实际值与预测值的拟合情况，实现交通预测模型的性能对比与效果可视化分析。

七、地铁数据预测
该页面为交通大数据分析系统的地铁数据预测模块，可选择目标城市，通过折线图区分真实值与预测值，直观呈现地铁历史客流量走势与未来一周的客流预测结果，支持数据刷新，实现地铁客流的趋势分析与未来预测的可视化展示。

八、交通数据预测
该页面为交通大数据分析系统的交通数据预测模块，可选择目标城市，通过折线图区分真实值与预测值，直观呈现城市历史高峰拥堵延时指数走势与未来一周的拥堵预测结果，支持数据刷新，实现交通拥堵趋势分析与未来预测的可视化展示。

九、登录页面
该页面为交通大数据分析系统的登录界面，提供账号与密码的输入区域，配备密码显示切换按钮与登录按钮，用于验证用户身份，实现系统的权限准入，保障系统数据安全，是进入系统各功能模块的入口。

3、项目说明

一、技术栈简要说明

本系统后端采用Python语言结合Flask轻量级框架构建，负责业务逻辑处理与API接口服务。前端使用Vue框架实现响应式用户界面，配合ECharts图表库完成数据可视化渲染。数据库选用MySQL存储爬虫采集的交通与地铁运营数据。数据采集环节通过Requests库从MetroDB网站定时抓取城市交通健康榜、客流量等公开信息。预测模块集成Arima统计模型与LSTM深度学习算法，分别用于时序趋势分析与非线性特征拟合。

二、功能模块详细介绍

• 首页仪表盘
  该页面为系统主控面板，顶部展示地铁城市总数、全网总客流量、平均健康指数等核心统计数据。中部采用柱状图呈现周一至周日的客流量分布规律，右侧饼图划分城市健康状态与亚健康状态的占比关系。底部折线图动态展示客流量前十城市的变化趋势，支持鼠标悬浮查看具体数值，实现全局数据多维度总览。

• 交通数据分析
  该模块包含中国主要城市交通健康榜与城市详情分析两个子模块。健康榜以表格形式列出各城市排名、健康指数、拥堵延时指数等指标，并辅以饼图展示健康状态分布。城市详情分析支持按近24小时或近7天时间范围筛选，可选择不同交通指标维度，通过折线图呈现拥堵指数时段变化，结合区域拥堵排名表格与热力地图，直观展示城市内部交通运行状态。

• 地铁数据分析
  该页面支持切换目标城市，通过面积图表展示地铁历史客流量的长期走势趋势，帮助用户识别客流高峰与低谷时段。页面下方提供分页表格，详细列出各城市不同时段的客流统计数据，支持按日期筛选查询，满足精细化数据挖掘与对比分析需求。

• 模型对比
  该模块支持选择目标城市，同时展示ARIMA与LSTM两种预测模型的核心评估指标，包括均方误差、平均绝对百分比误差等。通过两条折线图分别呈现两种模型的实际值与预测值的拟合情况，用户可直观对比统计模型与深度学习模型在同一数据集上的预测表现，为算法选型提供参考依据。

• 地铁数据预测
  该页面支持城市选择功能，折线图以不同颜色区分真实历史客流量与未来一周预测客流量。图表横轴为时间序列，纵轴为客流量数值，预测结果基于ARIMA与LSTM算法融合输出。页面配备数据刷新按钮，支持更新最新预测数据，实现地铁客流趋势分析与短期预判的可视化展示。

• 交通数据预测
  该模块针对城市高峰拥堵延时指数进行预测分析，支持选择目标城市。折线图清晰区分历史真实拥堵指数与未来一周预测拥堵指数，展示拥堵变化的时序规律。页面同样提供数据刷新功能，帮助交通管理部门提前掌握拥堵趋势，辅助运力调度与出行引导决策。

• 登录页面
  该界面为系统入口，提供账号与密码输入框，配备密码显示切换按钮，提升输入便捷性。用户提交凭证后系统进行身份验证，验证通过方可进入各功能模块。登录机制有效保障系统访问安全，防止未授权用户进入数据分析与预测界面。

三、项目总结

本系统围绕城市交通数据治理与预测需求，构建了从数据采集、存储、分析到预测的完整技术链路。通过Requests爬虫实现交通健康榜与地铁运营数据的自动化采集，经清洗后存入MySQL保障数据质量。七大功能模块协同工作，首页仪表盘提供全局概览，交通与地铁数据分析模块支持精细化多维度挖掘，模型对比模块辅助算法选型，地铁与交通数据预测模块输出未来一周趋势预测结果。可视化层面借助ECharts实现柱状图、饼图、折线图、面积图、热力地图等多种图表渲染，支持交互查询与数据刷新。预测算法融合Arima统计模型与LSTM神经网络，兼顾线性趋势与非线性特征。经测试验证，系统运行稳定，数据展示精准，预测误差控制在合理范围，可有效支持交通管理部门的拥堵疏导决策，同时为公众出行规划提供前瞻性参考。

4、核心代码

# -*- coding: UTF-8 -*-
"""
@summary:
@DATE: 11/15 17:53
"""

import asyncio
import csv

import aiohttp
import re
import json
# from tool.logger import logger
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
from bs4 import BeautifulSoup
import requests
from tqdm import tqdm
from typing import List, Any, Union

class DataFetcher:
    def __init__(self, base_url: str, headers: dict, cookies: dict):
        self.base_url = base_url
        self.headers = headers
        self.cookies = cookies

    async def get_response(self, session, url):
        """异步获取响应"""
        max_retries = 5  # 最大重试次数
        retries = 0
        while retries < max_retries:
            try:
                async with session.get(url, headers=self.headers, cookies=self.cookies) as response:
                    return await response.text()
            except aiohttp.ClientConnectorError as e:
                retries += 1
                print(f"连接错误，正在重试 {retries}/{max_retries}...")
                await asyncio.sleep(2)  # 等待2秒后重试
            except Exception as e:
                print(f"发生错误: {e}")
                break
        return None

    def is_null_number_string(self, string: str) -> Union[int, str]:
        """判断字符串是否为空数字字符串"""
        return 0 if string.strip() == "" else string

    def process_city_data(self, city, response_text, year_list):
        """处理每个城市的数据"""
        new_data = re.findall(r'data: \[(.*?)\]', response_text)
        years = new_data[1].split(', ')[:-1]
        flows = new_data[2].split(', ')[:-1]

        flow_list = [0] * len(year_list)
        for i, year in enumerate(years):
            year = int(year)
            if year in year_list:
                index = year_list.index(year)
                flow_list[index] = flows[i]

        new_data = re.findall(r'rollNum\(".*", 0, (.*?)\);', response_text)
        data = {
            'city': city[1],
            'time': new_data[0],
            'operating_line': new_data[1],
            'under_construction_line': new_data[2],
            'operating_mileage': self.is_null_number_string(new_data[3].replace(', 2', '')),
            'max_passenger_flow': self.is_null_number_string(new_data[4].replace(', 2', '')),
            'yesterday_passenger_flow': self.is_null_number_string(new_data[6].replace(', 2', '')),
            'yesterday_passenger_intensity': self.is_null_number_string(new_data[7].replace(', 2', '')),
            'operating_station_count': self.is_null_number_string(new_data[8]),
            'year_flow': json.dumps(dict(zip(year_list, flow_list)), ensure_ascii=False),
        }
        return data

    async def fetch_city_data(self, session, city, year_list):
        """异步获取每个城市的数据"""
        url = f"{self.base_url}/index/{city[0]}"
        while True:
            try:
                response_text = await self.get_response(session, url)
                return self.process_city_data(city, response_text, year_list)
            except aiohttp.ClientError:
                continue

    async def get_data(self, city_url: str, year_list: List[int], max_workers=10):
        """获取并处理数据"""
        async with aiohttp.ClientSession() as session:
            # 获取城市列表
            response_text = await self.get_response(session, city_url)
            city_list = re.findall('<a href="(.*?)">(.*?)</a>', response_text)[1:-1]

            total_cities = len(city_list)
            results = []
            errors = []

            # 使用线程池来加速写入等操作
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                with tqdm(total=total_cities, desc="处理城市数据") as pbar:
                    loop = asyncio.get_event_loop()
                    tasks = []

                    # 创建任务
                    for city in city_list:
                        tasks.append(self.fetch_city_data(session, city, year_list))

                    # 处理完成的任务
                    for future in asyncio.as_completed(tasks):
                        try:
                            city_data = await future
                            results.append(city_data)
                            pbar.update(1)
                        except Exception as e:
                            pbar.update(1)
                            errors.append(str(e))

            return results, errors

    def generate_date_lists(self, start_date, end_date):
        """根据日期生成年份列表"""
        start = datetime.strptime(start_date, '%Y-%m-%d')
        end = datetime.strptime(end_date, '%Y-%m-%d')
        return [year for year in range(start.year, end.year + 1)]


async def get_metro_db_data():
    # 初始化DataFetcher实例

    city_url = "https://www.metrodb.org/index/guangzhou.html"

    # 创建DataFetcher实例
    fetcher = DataFetcher(base_url="https://www.metrodb.org", headers=headers, cookies=cookies)

    # 获取年份列表
    year_list = fetcher.generate_date_lists('2018-01-01', datetime.now().strftime('%Y-%m-%d'))

    # 获取数据并返回
    result, errors = await fetcher.get_data(city_url, year_list, max_workers=10)
    if errors:
        print(f"处理过程中发生错误：{errors}")
    return result


def get_response(url, headers, method: str = "GET", cookies: dict = None):
    response = None
    if method == "GET":
        try:
            response = requests.get(url, headers=headers, cookies=cookies, timeout=10)
        except:
            max_retries = 5  # 最大重试次数
            retries = 1
            while retries < max_retries:
                print(f"连接错误，正在重试 {retries}/{max_retries}...")
                try:
                    response = requests.get(url, headers=headers, cookies=cookies, timeout=10)

                    return response
                except:
                    retries += 1
    else:
        response = requests.post(url, headers=headers, cookies=cookies, timeout=10)
    return response


def to_date(date_str):
    return datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S")


def to_timestamp(timestamp):
    # 将时间戳转换为datetime对象
    dt_object = datetime.fromtimestamp(timestamp)

    # 将datetime对象格式化为字符串
    return dt_object.strftime("%Y-%m-%d %H:%M:%S")


def get_metro_watch_data():


def get_real_time_traffic_detail_data(code, data_type: str):
    """
    @summary: 获取实时交通详情数据
    """
    url = "https://report.amap.com/ajax/cityHourly.do"
    params = {
        "cityCode": code,
        "dataType": data_type
    }
    response = requests.get(url, headers=headers, params=params, cookies=cookies)
    json_data = response.json()
    data_list = []
    print(type(data_type))
    if data_type != '5':
        for item in json_data:
            data_list.append([to_timestamp(item[0] / 1000), item[1]])
        return data_list
    else:
        return json_data


def get_real_time_traffic_detail_data2(code, data_type: str):
    """
    @summary: 获取实时交通详情数据
    """
    url = "https://report.amap.com/ajax/cityDaily.do"
    params = {
        "cityCode": code,
        "dataType": data_type
    }
    response = requests.get(url, headers=headers, params=params, cookies=cookies)
    json_data = response.json()
    data_list = []
    if data_type != '5':
        for item in json_data:
            data_list.append([to_timestamp(item[0] / 1000), item[1]])
        return data_list
    else:
        return json_data


async def get_traffic_data():
    url =
    response = get_response(url, headers, cookies=cookies)
    json_data = response.json()
    city_traffic = []
    for item in tqdm(json_data):
        code = item['adcode']
        rank = item['rankchange1']
        time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        cityName = item['cityName']
        healthValue = item['healthValue']
        idx = item['idx']
        realSpeed = item['realSpeed']
        city_traffic.append({
            'rank': rank,
            'city': cityName,
            'city_code': code,
            'timestamp': time,
            'traffic_health_index': healthValue,
            'congestion_delay_index': idx,
            'average_speed': realSpeed
        })
    return city_traffic


if __name__ == "__main__":
    print(to_timestamp(1733533200))
    get_metro_watch_data()
    ls = get_real_time_traffic_detail_data('321200', '1')   # 获取实时交通详情数据
    print(ls)
    pass

5、项目列表