Streamlit MySQL Plotly 疫情数据可视化大屏一、项目概述构建一个实时更新的疫情数据可视化大屏支持全球/中国各省疫情数据的采集、存储、分析与交互式展示。技术栈前端框架Streamlit快速搭建数据应用数据库MySQL 8.x结构化存储可视化引擎Plotly交互式图表数据源丁香园 / 国家卫健委公开数据二、系统架构数据采集层 存储层 展示层 丁香园API ──▶ collector.py ──▶ MySQL ──▶ dashboard.py (Streamlit) │ ├── 全国趋势图 (Plotly) ├── 各省热力图 (Plotly) ├── 实时排名表 ├── 关键指标卡片 └── 预测模型 (可选)三、MySQL 表结构设计CREATE DATABASE covid19_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE covid19_db; -- 全国每日统计数据 CREATE TABLE daily_stats ( id BIGINT AUTO_INCREMENT PRIMARY KEY, report_date DATE NOT NULL UNIQUE, confirmed INT DEFAULT 0 COMMENT 累计确诊, suspected INT DEFAULT 0 COMMENT 疑似, cured INT DEFAULT 0 COMMENT 治愈, dead INT DEFAULT 0 COMMENT 死亡, imported_case INT DEFAULT 0 COMMENT 境外输入, asymptomatic INT DEFAULT 0 COMMENT 无症状, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME ON UPDATE CURRENT_TIMESTAMP ) ENGINEInnoDB COMMENT全国每日疫情统计; -- 各省份每日数据 CREATE TABLE province_stats ( id BIGINT AUTO_INCREMENT PRIMARY KEY, report_date DATE NOT NULL, province VARCHAR(50) NOT NULL COMMENT 省份名称, confirmed INT DEFAULT 0, suspected INT DEFAULT 0, cured INT DEFAULT 0, dead INT DEFAULT 0, confirmed_add INT DEFAULT 0 COMMENT 新增确诊, cured_add INT DEFAULT 0 COMMENT 新增治愈, dead_add INT DEFAULT 0 COMMENT 新增死亡, UNIQUE KEY uk_date_province (report_date, province), INDEX idx_province (province), INDEX idx_date (report_date) ) ENGINEInnoDB COMMENT各省每日疫情数据; -- 全球各国数据 CREATE TABLE global_stats ( id BIGINT AUTO_INCREMENT PRIMARY KEY, report_date DATE NOT NULL, country VARCHAR(100) NOT NULL COMMENT 国家, confirmed INT DEFAULT 0, cured INT DEFAULT 0, dead INT DEFAULT 0, confirmed_add INT DEFAULT 0, UNIQUE KEY uk_date_country (report_date, country), INDEX idx_country (country) ) ENGINEInnoDB COMMENT全球各国疫情数据;四、数据采集模块4.1 数据源配置# config.py import os from dataclasses import dataclass dataclass class DBConfig: host: str os.getenv(MYSQL_HOST, localhost) port: int int(os.getenv(MYSQL_PORT, 3306)) user: str os.getenv(MYSQL_USER, root) password: str os.getenv(MYSQL_PASSWORD, password) database: str covid19_db property def connection_string(self): return fmysqlpymysql://{self.user}:{self.password}{self.host}:{self.port}/{self.database}?charsetutf8mb44.2 数据采集器# collector.py import requests import pandas as pd from datetime import datetime, timedelta from sqlalchemy import create_engine import time import logging from config import DBConfig logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class COVIDCollector: def __init__(self): self.engine create_engine(DBConfig().connection_string) def fetch_national_data(self): 采集全国数据模拟丁香园API try: # 实际项目中替换为真实API url https://lab.isaaclin.cn/nCoV/api/overall resp requests.get(url, timeout10) data resp.json()[results][0] df pd.DataFrame([{ report_date: datetime.now().date(), confirmed: data.get(confirmedCount, 0), suspected: data.get(suspectedCount, 0), cured: data.get(curedCount, 0), dead: data.get(deadCount, 0), imported_case: data.get(importedCaseCount, 0), asymptomatic: data.get(asymptomaticCount, 0) }]) df.to_sql(daily_stats, self.engine, if_existsappend, indexFalse) logger.info(f全国数据采集成功: {df.iloc[0][confirmed]}例) return True except Exception as e: logger.error(f全国数据采集失败: {e}) return False def fetch_province_data(self): 采集各省数据 try: url https://lab.isaaclin.cn/nCoV/api/area?latest1 resp requests.get(url, timeout10) results resp.json()[results] records [] today datetime.now().date() for province in results: records.append({ report_date: today, province: province.get(provinceShortName, ), confirmed: province.get(confirmedCount, 0), suspected: province.get(suspectedCount, 0), cured: province.get(curedCount, 0), dead: province.get(deadCount, 0), confirmed_add: province.get(confirmedCount, 0) - province.get(cachedConfirmedCount, 0), cured_add: province.get(curedCount, 0) - province.get(cachedCuredCount, 0), dead_add: province.get(deadCount, 0) - province.get(cachedDeadCount, 0) }) df pd.DataFrame(records) df.to_sql(province_stats, self.engine, if_existsappend, indexFalse) logger.info(f各省数据采集成功: {len(records)}个省/地区) return True except Exception as e: logger.error(f各省数据采集失败: {e}) return False def mock_data_generator(self): 生成模拟数据演示用 today datetime.now().date() provinces [北京, 上海, 广东, 浙江, 江苏, 湖北, 四川, 山东, 河南, 福建] np.random.seed(42) # 生成全国数据 national_data pd.DataFrame({ report_date: pd.date_range(endtoday, periods90), confirmed: np.cumsum(np.random.randint(100, 1500, 90)), suspected: np.random.randint(100, 500, 90), cured: np.cumsum(np.random.randint(50, 900, 90)), dead: np.cumsum(np.random.randint(5, 15, 90)), imported_case: np.random.randint(10, 50, 90), asymptomatic: np.random.randint(50, 200, 90) }) national_data.to_sql(daily_stats, self.engine, if_existsreplace, indexFalse) # 生成各省数据 province_records [] for date in pd.date_range(endtoday, periods30): for prov in provinces: base_confirmed np.random.randint(1000, 10000) province_records.append({ report_date: date, province: prov, confirmed: base_confirmed, suspected: np.random.randint(50, 200), cured: int(base_confirmed * np.random.uniform(0.85, 0.95)), dead: np.random.randint(5, 50), confirmed_add: np.random.randint(0, 100), cured_add: np.random.randint(50, 200), dead_add: np.random.randint(0, 5) }) df_province pd.DataFrame(province_records) df_province.to_sql(province_stats, self.engine, if_existsreplace, indexFalse) logger.info(模拟数据生成完成) return True if __name__ __main__: collector COVIDCollector() collector.mock_data_generator() # 首次运行生成模拟数据五、Streamlit 大屏应用# dashboard.py import streamlit as st import pandas as pd import plotly.express as px import plotly.graph_objects as go from plotly.subplots import make_subplots from sqlalchemy import create_engine from datetime import datetime, timedelta import numpy as np from config import DBConfig # 页面配置 st.set_page_config( page_title疫情数据可视化大屏, page_icon, layoutwide, initial_sidebar_stateexpanded ) # 数据库连接 st.cache_resource def get_engine(): return create_engine(DBConfig().connection_string) engine get_engine() # 数据加载函数 st.cache_data(ttl300) def load_national_data(): query SELECT * FROM daily_stats ORDER BY report_date DESC LIMIT 365 df pd.read_sql(query, engine, parse_dates[report_date]) return df.sort_values(report_date) st.cache_data(ttl300) def load_province_latest(): query SELECT p1.* FROM province_stats p1 INNER JOIN ( SELECT province, MAX(report_date) as max_date FROM province_stats GROUP BY province ) p2 ON p1.province p2.province AND p1.report_date p2.max_date return pd.read_sql(query, engine) st.cache_data(ttl300) def load_province_trend(province, days30): query f SELECT * FROM province_stats WHERE province {province} ORDER BY report_date DESC LIMIT {days} df pd.read_sql(query, engine, parse_dates[report_date]) return df.sort_values(report_date) # 侧边栏 st.sidebar.title( 疫情数据大屏) st.sidebar.markdown(---) st.sidebar.header(数据筛选) # 时间范围选择 today datetime.now().date() date_range st.sidebar.date_input( 选择时间范围, value(today - timedelta(days30), today), max_valuetoday ) if len(date_range) 2: start_date, end_date date_range else: start_date today - timedelta(days30) end_date today # 省份选择 province_data load_province_latest() provinces [全国] sorted(province_data[province].unique().tolist()) selected_province st.sidebar.selectbox(选择省份, provinces) st.sidebar.markdown(---) st.sidebar.info(数据来源模拟数据 | 更新时间每日10:00) # 主面板 st.title( 新冠疫情数据可视化大屏) st.markdown(f**数据范围**: {start_date} 至 {end_date}) # 加载数据 national_data load_national_data() # 获取最新数据 latest_national national_data.iloc[-1] if not national_data.empty else None # 1️⃣ 关键指标卡片 st.markdown(## 核心指标) col1, col2, col3, col4, col5 st.columns(5) if latest_national is not None: with col1: st.metric( label累计确诊, valuef{latest_national[confirmed]:,}, deltaf{national_data.iloc[-1][confirmed] - national_data.iloc[-2][confirmed]} if len(national_data) 1 else None, delta_colorinverse ) with col2: st.metric( label现有确诊, valuef{latest_national[confirmed] - latest_national[cured] - latest_national[dead]:,}, delta_colorinverse ) with col3: st.metric( label累计治愈, valuef{latest_national[cured]:,}, deltaf{latest_national[cured] - national_data.iloc[-2][cured]} if len(national_data) 1 else None ) with col4: st.metric( label累计死亡, valuef{latest_national[dead]:,}, deltaf{latest_national[dead] - national_data.iloc[-2][dead]} if len(national_data) 1 else None, delta_colorinverse ) with col5: mortality_rate (latest_national[dead] / latest_national[confirmed] * 100) if latest_national[confirmed] 0 else 0 st.metric( label死亡率, valuef{mortality_rate:.2f}%, delta_colorinverse ) st.markdown(---) # 2️⃣ 趋势图 st.markdown(## 疫情发展趋势) fig make_subplots( rows2, cols2, subplot_titles(确诊与治愈趋势, 新增趋势, 每日新增确诊, 关键指标对比), vertical_spacing0.12, horizontal_spacing0.1 ) # 过滤日期范围 mask (national_data[report_date] pd.Timestamp(start_date)) \ (national_data[report_date] pd.Timestamp(end_date)) filtered_data national_data[mask].copy() if not filtered_data.empty: # 子图1: 确诊与治愈趋势 fig.add_trace( go.Scatter(xfiltered_data[report_date], yfiltered_data[confirmed], modelines, name累计确诊, linedict(color#FF6B6B, width2)), row1, col1 ) fig.add_trace( go.Scatter(xfiltered_data[report_date], yfiltered_data[cured], modelines, name累计治愈, linedict(color#51CF66, width2)), row1, col1 ) fig.add_trace( go.Scatter(xfiltered_data[report_date], yfiltered_data[dead], modelines, name累计死亡, linedict(color#868E96, width2)), row1, col1 ) # 计算每日新增 filtered_data[new_confirmed] filtered_data[confirmed].diff().fillna(0) filtered_data[new_cured] filtered_data[cured].diff().fillna(0) # 子图2: 新增趋势 fig.add_trace( go.Bar(xfiltered_data[report_date], yfiltered_data[new_confirmed], name新增确诊, marker_color#FF6B6B), row1, col2 ) fig.add_trace( go.Bar(xfiltered_data[report_date], yfiltered_data[new_cured], name新增治愈, marker_color#51CF66), row1, col2 ) # 子图3: 每日新增确诊柱状图 fig.add_trace( go.Bar(xfiltered_data[report_date], yfiltered_data[new_confirmed], name每日新增, marker_color#FF8787), row2, col1 ) # 子图4: 多指标对比 fig.add_trace( go.Scatter(xfiltered_data[report_date], yfiltered_data[suspected], modelinesmarkers, name疑似病例, linedict(color#FFD43B, width2)), row2, col2 ) fig.add_trace( go.Scatter(xfiltered_data[report_date], yfiltered_data[imported_case], modelinesmarkers, name境外输入, linedict(color#74C0FC, width2)), row2, col2 ) fig.add_trace( go.Scatter(xfiltered_data[report_date], yfiltered_data[asymptomatic], modelinesmarkers, name无症状, linedict(color#DA77F2, width2)), row2, col2 ) fig.update_layout(height700, showlegendTrue, templateplotly_white) fig.update_xaxes(title_text日期) fig.update_yaxes(title_text人数) st.plotly_chart(fig, use_container_widthTrue) st.markdown(---) # 3️⃣ 各省数据展示 st.markdown(## ️ 各省疫情分布) col1, col2 st.columns([2, 1]) with col1: # 各省确诊热力图 province_latest load_province_latest() fig_map px.choropleth( province_latest, locationsprovince, locationmodechina, colorconfirmed, hover_nameprovince, hover_data{ confirmed: :, , cured: :, , dead: :, , confirmed_add: :, }, color_continuous_scaleReds, title各省累计确诊分布, height500 ) fig_map.update_layout( geodict( projectiondict(typemercator), scopeasia, showframeFalse, showcountriesTrue, countrycolorrgb(204, 204, 204) ) ) st.plotly_chart(fig_map, use_container_widthTrue) with col2: # 各省确诊排名 top_provinces province_latest.nlargest(10, confirmed)[[province, confirmed, cured, dead]] fig_bar px.bar( top_provinces, xconfirmed, yprovince, orientationh, title确诊TOP10省份, labels{confirmed: 确诊人数, province: }, colorconfirmed, color_continuous_scaleReds, height500 ) fig_bar.update_layout(yaxis{categoryorder: total ascending}) st.plotly_chart(fig_bar, use_container_widthTrue) st.markdown(---) # 4️⃣ 省份详细分析 if selected_province ! 全国: st.markdown(f## {selected_province}疫情详细分析) province_trend load_province_trend(selected_province, days60) if not province_trend.empty: col1, col2 st.columns(2) with col1: fig_province go.Figure() fig_province.add_trace(go.Scatter( xprovince_trend[report_date], yprovince_trend[confirmed], modelines, name累计确诊, linedict(color#FF6B6B, width3) )) fig_province.add_trace(go.Scatter( xprovince_trend[report_date], yprovince_trend[cured], modelines, name累计治愈, linedict(color#51CF66, width3) )) fig_province.add_trace(go.Scatter( xprovince_trend[report_date], yprovince_trend[dead], modelines, name累计死亡, linedict(color#868E96, width3) )) fig_province.update_layout( titlef{selected_province}疫情趋势, xaxis_title日期, yaxis_title人数, height400 ) st.plotly_chart(fig_province, use_container_widthTrue) with col2: # 新增趋势 province_trend[new_confirmed] province_trend[confirmed_add] fig_new go.Figure() fig_new.add_trace(go.Bar( xprovince_trend[report_date], yprovince_trend[new_confirmed], name新增确诊, marker_color#FF6B6B )) fig_new.add_trace(go.Bar( xprovince_trend[report_date], yprovince_trend[cured_add], name新增治愈, marker_color#51CF66 )) fig_new.update_layout( titlef{selected_province}每日新增, xaxis_title日期, yaxis_title人数, barmodegroup, height400 ) st.plotly_chart(fig_new, use_container_widthTrue) # 5️⃣ 数据表格 st.markdown(---) st.markdown(## 原始数据预览) show_data st.checkbox(显示原始数据, valueFalse) if show_data: if selected_province 全国: st.dataframe(filtered_data.style.highlight_max(axis0), use_container_widthTrue) else: st.dataframe(province_trend.style.highlight_max(axis0), use_container_widthTrue) # 底部信息 st.markdown(---) st.markdown( div styletext-align: center; color: #868E96; padding: 20px; p数据来源模拟数据演示用途 | 更新时间每日10:00/p p技术支持Streamlit MySQL Plotly/p /div , unsafe_allow_htmlTrue)六、运行说明6.1 启动脚本# 1. 安装依赖 pip install streamlit pandas numpy plotly sqlalchemy pymysql requests # 2. 初始化数据库 mysql -u root -p schema.sql # 3. 生成模拟数据 python collector.py # 4. 启动大屏 streamlit run dashboard.py --server.port 85016.2 Docker 部署# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8501 CMD [streamlit, run, dashboard.py, --server.port8501, --server.address0.0.0.0]# docker-compose.yml version: 3.8 services: mysql: image: mysql:8.0 environment: MYSQL_ROOT_PASSWORD: password MYSQL_DATABASE: covid19_db ports: - 3306:3306 volumes: - ./schema.sql:/docker-entrypoint-initdb.d/schema.sql - mysql_data:/var/lib/mysql app: build: . ports: - 8501:8501 depends_on: - mysql environment: MYSQL_HOST: mysql MYSQL_USER: root MYSQL_PASSWORD: password volumes: mysql_data:七、功能特色功能模块实现方式交互特点核心指标卡Streamlit metric组件实时Delta变化颜色预警趋势分析Plotly多子图缩放、悬停、框选地理分布Plotly中国地图色阶映射悬浮详情省份排名横向柱状图自动排序颜色渐变省份钻取联动筛选点击侧边栏切换数据缓存st.cache_data5分钟自动刷新
