当前位置：首页 > news >正文

194-基于Python的脑肿瘤患者数据分析可视化

news 2025/9/24 10:33:17

🧠 基于Python的脑肿瘤患者数据分析可视化系统：从数据到智能预测的完整解决方案

码界筑梦坊 | 2025年技术分享

📋 目录

项目概述
技术架构
核心功能实现
数据可视化展示
机器学习模型
代码实现详解
部署与运行
性能优化
项目总结
联系方式

项目概述

在医疗数据科学领域，脑肿瘤数据分析是一个极具挑战性的课题。本项目基于Python构建了一个完整的脑肿瘤患者数据分析可视化系统，集成了数据管理、多维度分析、智能预测和现代化Web界面，为医疗机构和研究人员提供了强大的数据分析工具。

🎯 项目亮点

全栈技术栈：FastAPI + SQLite + ECharts + Bootstrap
智能预测：基于机器学习的肿瘤类型分类和存活率预测
丰富可视化：多维度数据分析和交互式图表展示
现代化界面：响应式设计，支持多设备访问
完整功能：从数据管理到AI预测的全流程解决方案

技术架构

🏗️ 整体架构图

🛠️ 技术栈详解

后端技术

Python 3.8+ - 主要编程语言
FastAPI - 现代化异步Web框架
Uvicorn - 高性能ASGI服务器
SQLite - 轻量级关系型数据库
aiosqlite - 异步数据库操作

数据处理

Pandas - 数据分析和处理
NumPy - 数值计算
Scikit-learn - 机器学习算法库
Joblib - 模型持久化

前端技术

HTML5/CSS3 - 页面结构
Bootstrap 5 - 响应式UI框架
ECharts - 数据可视化图表库
jQuery - JavaScript工具库

机器学习

RandomForest - 随机森林算法
特征工程 - 自动数据预处理
模型评估 - 性能指标计算

项目演示

在这里插入图片描述

核心功能实现

🔐 用户认证系统

# 用户认证核心代码
from fastapi import FastAPI, Request, HTTPException, status, Form
from starlette.middleware.sessions import SessionMiddlewareapp.add_middleware(SessionMiddleware, secret_key="brain-tumor-analysis-secret-key-2024")async def authenticate_user(self, username: str, password: str) -> dict:"""用户认证"""async with aiosqlite.connect(self.db_path) as db:password_hash = self.hash_password(password)cursor = await db.execute("SELECT id, username, email FROM users WHERE username = ? AND password_hash = ?",(username, password_hash))row = await cursor.fetchone()if row:return {"id": row[0],"username": row[1],"email": row[2]}return None

📊 数据管理模块

# 患者数据管理
async def get_patients(self, page: int = 1, size: int = 20, search: str = ""):"""获取患者数据（分页）"""async with aiosqlite.connect(self.db_path) as db:db.row_factory = aiosqlite.Rowoffset = (page - 1) * sizewhere_clause = ""params = []if search:# 支持多条件搜索search_parts = search.strip().split()for part in search_parts:if ':' in part:key, value = part.split(':', 1)if key in ['gender', 'tumor_type', 'location']:where_clause += f" AND {key} = ?"params.append(value)else:where_clause += " AND (patient_id LIKE ? OR age LIKE ?)"params.extend([f"%{part}%", f"%{part}%"])query = f"""SELECT * FROM brain_tumor_patients WHERE 1=1 {where_clause}ORDER BY created_at DESC LIMIT ? OFFSET ?"""params.extend([size, offset])cursor = await db.execute(query, params)patients = await cursor.fetchall()# 获取总数count_query = f"SELECT COUNT(*) FROM brain_tumor_patients WHERE 1=1 {where_clause}"count_cursor = await db.execute(count_query, params[:-2])total = await count_cursor.fetchone()return {"patients": [dict(patient) for patient in patients],"total": total[0],"page": page,"size": size,"pages": (total[0] + size - 1) // size}

📈 数据分析模块

# 肿瘤类型分析
async def get_tumor_type_analysis(self):"""肿瘤类型分析"""data = await self.db.get_all_patients_data()df = pd.DataFrame(data)# 统一数据格式df['gender'] = df['gender'].replace({'男': 'Male', '女': 'Female'})df['tumor_type'] = df['tumor_type'].replace({'良性': 'Benign', '恶性': 'Malignant'})# 基础统计malignant_count = len(df[df['tumor_type'] == 'Malignant'])benign_count = len(df[df['tumor_type'] == 'Benign'])total_count = len(df)statistics = {'malignant_count': malignant_count,'benign_count': benign_count,'total_count': total_count,'malignant_rate': round((malignant_count / total_count * 100), 1),'benign_rate': round((benign_count / total_count * 100), 1)}# 性别与肿瘤类型关系gender_tumor = df.groupby(['gender', 'tumor_type']).size().unstack(fill_value=0)# 年龄段分析age_bins = [20, 30, 40, 50, 60, 70]age_labels = ['20-30岁', '30-40岁', '40-50岁', '50-60岁', '60-70岁']df['age_group'] = pd.cut(df['age'], bins=age_bins, labels=age_labels, right=False)return self.clean_data_for_json({'statistics': statistics,'gender_analysis': gender_tumor.to_dict(),'age_analysis': df.groupby(['age_group', 'tumor_type']).size().unstack(fill_value=0).to_dict()})

数据可视化展示

📊 图表类型与实现

1. 肿瘤类型分布 - 水球图

// ECharts水球图配置
const tumorTypeOption = {series: [{type: 'liquidFill',data: [malignantRate / 100],color: ['#ff6b6b'],center: ['50%', '50%'],radius: '80%',label: {normal: {textStyle: {color: '#fff',fontSize: 20,fontWeight: 'bold'}}}}]
};

2. 年龄分布分析 - 柱状图

// 年龄分布柱状图
const ageDistributionOption = {title: {text: '患者年龄分布',left: 'center'},tooltip: {trigger: 'axis',axisPointer: {type: 'shadow'}},xAxis: {type: 'category',data: ageLabels},yAxis: {type: 'value'},series: [{name: '患者数量',type: 'bar',data: ageCounts,itemStyle: {color: '#4ecdc4'}}]
};

3. 治疗效果分析 - 雷达图

// 治疗效果雷达图
const treatmentOption = {radar: {indicator: [{ name: '手术治疗', max: 100 },{ name: '放射治疗', max: 100 },{ name: '化学治疗', max: 100 },{ name: '综合治疗', max: 100 }]},series: [{name: '治疗效果',type: 'radar',data: [{value: treatmentData,name: '治疗方式对比'}]}]
};

🎨 可视化效果展示

注意：以下为预留的可视化展示区域，实际项目中包含丰富的交互式图表

仪表板概览

📈 实时统计数据展示
🎯 关键指标监控
📊 多维度数据对比

分析图表

🥧 肿瘤类型分布饼图
📊 年龄分布柱状图
🌡️ 存活率趋势图
🎯 治疗效果雷达图
📍 位置分布地图

机器学习模型

🤖 模型架构

class BrainTumorModel:def __init__(self):self.tumor_type_model = Noneself.survival_rate_model = Noneself.label_encoders = {}self.scaler = StandardScaler()self.is_trained = False

🔧 特征工程

def preprocess_data(self, df):"""数据预处理"""data = df.copy()# 处理分类变量categorical_features = ['Gender', 'Location', 'Histology', 'Stage', 'Radiation_Treatment', 'Surgery_Performed', 'Chemotherapy', 'Family_History', 'MRI_Result']for feature in categorical_features:if feature not in self.label_encoders:self.label_encoders[feature] = LabelEncoder()data[feature] = self.label_encoders[feature].fit_transform(data[feature])return data

🎯 模型训练

async def train_tumor_type_classifier(self, data):"""训练肿瘤类型分类器"""features = ['Age', 'Gender', 'Tumor_Size', 'Location', 'Histology', 'Stage', 'Family_History', 'MRI_Result']X = data[features]y = data['Tumor_Type_Encoded']# 数据分割X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)# 特征标准化X_train_scaled = self.scaler.fit_transform(X_train)X_test_scaled = self.scaler.transform(X_test)# 训练随机森林模型self.tumor_type_model = RandomForestClassifier(n_estimators=100, random_state=42)self.tumor_type_model.fit(X_train_scaled, y_train)# 模型评估y_pred = self.tumor_type_model.predict(X_test_scaled)accuracy = accuracy_score(y_test, y_pred)print(f"肿瘤类型分类模型准确率: {accuracy:.4f}")

🔮 智能预测

async def predict(self, patient_data):"""进行预测"""try:# 预处理输入数据processed_input = self.preprocess_input(patient_data)# 肿瘤类型预测tumor_features = ['Age', 'Gender', 'Tumor_Size', 'Location', 'Histology', 'Stage', 'Family_History', 'MRI_Result']tumor_input = [processed_input.get(feature, 0) for feature in tumor_features]tumor_input_scaled = self.scaler.transform([tumor_input])tumor_pred = self.tumor_type_model.predict(tumor_input_scaled)[0]tumor_prob = self.tumor_type_model.predict_proba(tumor_input_scaled)[0]# 存活率预测survival_features = ['Age', 'Gender', 'Tumor_Size', 'Location', 'Histology', 'Stage', 'Radiation_Treatment', 'Surgery_Performed', 'Chemotherapy', 'Tumor_Growth_Rate', 'Family_History']survival_input = [processed_input.get(feature, 0) for feature in survival_features]survival_rate = self.survival_rate_model.predict([survival_input])[0]return {"tumor_type": self.label_encoders['Tumor_Type'].inverse_transform([tumor_pred])[0],"tumor_type_probability": float(max(tumor_prob)),"predicted_survival_rate": float(survival_rate),"risk_level": self.get_risk_level(tumor_pred, survival_rate)}except Exception as e:return {"error": str(e)}

代码实现详解

📁 项目结构

brain-tumor-analysis/
├── main.py                 # 主应用程序入口
├── database.py             # 数据库操作模块
├── data_analyzer.py        # 数据分析模块
├── ml_model.py            # 机器学习模型
├── retrain_models.py      # 模型重训练脚本
├── init_database.py       # 数据库初始化
├── start.py               # 启动脚本
├── requirements.txt       # 依赖包列表
├── brain_tumor_analysis.db # SQLite数据库文件
├── models/                # 机器学习模型文件
│   ├── survival_rate_model.pkl
│   ├── tumor_type_model.pkl
│   ├── scaler.pkl
│   └── label_encoders.pkl
├── data/                  # 数据文件
│   └── brain_tumor_dataset.csv
├── templates/             # HTML模板
│   ├── base.html
│   ├── login.html
│   ├── register.html
│   ├── dashboard.html
│   ├── data_management.html
│   ├── analysis.html
│   ├── prediction.html
│   ├── profile.html
│   └── analysis/
│       ├── tumor_type.html
│       ├── age_distribution.html
│       ├── treatment_effectiveness.html
│       ├── symptoms.html
│       └── location.html
├── static/                # 静态资源
│   ├── css/
│   ├── js/
│   ├── fonts/
│   └── image/
└── README.md             # 项目说明文档

🚀 启动流程

# main.py - 应用启动
@asynccontextmanager
async def lifespan(app: FastAPI):# 启动时初始化global db, ml_model, data_analyzer# 初始化核心组件db = Database()ml_model = BrainTumorModel()data_analyzer = DataAnalyzer()# 初始化数据库await db.init_database()# 创建默认管理员用户await create_default_user()# 加载数据到数据库await db.load_brain_tumor_data()# 训练机器学习模型await ml_model.train_model()print("系统初始化完成!")yield# 关闭时清理资源print("系统正在关闭...")

🔧 核心API接口

# 数据分析API
@app.get("/api/analysis/tumor-type")
async def get_tumor_type_analysis(request: Request):"""获取肿瘤类型分析数据"""redirect_response = require_login(request)if redirect_response:return redirect_responseanalysis_data = await data_analyzer.get_tumor_type_analysis()return JSONResponse(content=analysis_data)# AI预测API
@app.post("/api/predict")
async def predict_tumor(request: Request, patient_data: dict):"""AI预测接口"""redirect_response = require_login(request)if redirect_response:return redirect_responseprediction = await ml_model.predict(patient_data)return JSONResponse(content=prediction)# 患者数据API
@app.get("/api/patients")
async def get_patients(request: Request,page: int = 1,size: int = 20,search: str = ""
):"""获取患者数据"""redirect_response = require_login(request)if redirect_response:return redirect_responsepatients_data = await db.get_patients(page, size, search)return JSONResponse(content=patients_data)

部署与运行

📦 环境配置

# 创建虚拟环境
python -m venv venv# 激活虚拟环境
# Windows
venv\Scripts\activate
# Linux/Mac
source venv/bin/activate# 安装依赖
pip install -r requirements.txt

🚀 启动服务

# 初始化数据库
python init_database.py# 启动应用
python start.py
# 或
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

🌐 访问地址

系统首页: http://localhost:8000
API文档: http://localhost:8000/docs
默认账户: admin / password

性能优化

⚡ 数据库优化

# 数据库索引优化
async def init_database(self):"""初始化数据库表结构"""async with aiosqlite.connect(self.db_path) as db:# 创建索引await db.execute("""CREATE INDEX IF NOT EXISTS idx_patients_gender ON brain_tumor_patients(gender)""")await db.execute("""CREATE INDEX IF NOT EXISTS idx_patients_tumor_type ON brain_tumor_patients(tumor_type)""")await db.execute("""CREATE INDEX IF NOT EXISTS idx_patients_age ON brain_tumor_patients(age)""")

🚀 缓存机制

# 数据缓存
from functools import lru_cache@lru_cache(maxsize=128)
def get_cached_analysis_data(analysis_type: str):"""缓存分析数据"""# 实现缓存逻辑pass

📊 异步处理

# 异步数据处理
async def process_large_dataset(self, data):"""异步处理大数据集"""# 分批处理数据batch_size = 1000for i in range(0, len(data), batch_size):batch = data[i:i + batch_size]await self.process_batch(batch)