🧠 基于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 SessionMiddleware

app.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.Row
        
        offset = (page - 1) * size
        where_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 = None
        self.survival_rate_model = None
        self.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_response
    
    analysis_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_response
    
    prediction = 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_response
    
    patients_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 = 1000
    for i in range(0, len(data), batch_size):
        batch = data[i:i + batch_size]
        await self.process_batch(batch)

---

项目总结

🎯 技术亮点

1. 现代化技术栈: 采用FastAPI + SQLite + ECharts的现代化技术组合
2. 完整功能体系: 从数据管理到AI预测的全流程解决方案
3. 丰富可视化: 多维度数据分析和交互式图表展示
4. 智能预测: 基于机器学习的肿瘤类型分类和存活率预测
5. 用户体验: 响应式设计，支持多设备访问

📈 应用价值

• 医疗研究: 为脑肿瘤研究提供数据分析工具
• 临床辅助: 支持医生进行数据驱动的决策
• 教学演示: 可作为医疗数据科学的教学案例
• 技术学习: 展示全栈开发和机器学习的最佳实践

🔮 未来展望

• 算法优化: 集成更多机器学习算法
• 功能扩展: 增加更多分析维度和预测模型
• 性能提升: 优化大数据处理能力
• 移动端: 开发移动端应用

💡 技术收获

通过这个项目，我们深入实践了：

• 全栈开发: 前后端分离的Web应用开发
• 数据处理: Pandas和NumPy在数据分析中的应用
• 机器学习: Scikit-learn在医疗数据上的应用
• 数据可视化: ECharts在Web端的图表展示
• 系统设计: 模块化、可扩展的系统架构

---

联系方式

码界筑梦坊 - 专注技术分享与创新

• GitHub: 码界筑梦坊
• 小红书: 码界筑梦坊
• CSDN: 码界筑梦坊
• 知乎: 码界筑梦坊
• B站: 码界筑梦坊

---

感谢阅读！ 如果这篇文章对您有帮助，请给个⭐️支持一下！

本文基于实际项目代码编写，所有代码示例均经过测试验证。项目仅用于教育和研究目的，不应用于临床诊断。

---

最后更新时间：2025年9月