简介：这是一份面向数据科学初学者与机器学习实践者的中风风险预测建模项目，基于Kaggle公开医疗数据集，聚焦于利用年龄、性别、高血压、吸烟史等临床特征构建高判别力的二分类模型。资源以Jupyter Notebook为核心载体，完整覆盖数据清洗、缺失值处理（含吸烟状态特殊标注说明）、特征工程、多模型对比（如逻辑回归、随机森林）、AUC与F1分数评估及模型持久化全流程，目标直指高鲁棒性临床辅助预测能力。压缩包共11个文件，含3个核心Notebook（含详细分析与部署脚本）、1个CSV原始数据集、1个.pkl模型文件、1个heroku_deploy.py部署脚本、1个Readme.md说明文档及配套Shell脚本与图像素材，整体4.36MB，结构紧凑、开箱即用。已有742人学习下载，读者可直接复现端到端建模流程，获取可部署的预测模型、标准化预处理代码、可视化分析片段及Heroku上线参考方案，特别适合课程设计、竞赛备赛与医疗AI入门实战。

1. 这不是另一个“准确率99%”的模型：一个真正能跑在基层医院检验科的脑卒中风险预测模块

某三甲医院神经内科合作项目里，我们把十几个“高分模型”全推倒重来——不是因为它们不准，而是因为部署后没人敢用。医生反馈：“模型说张大爷有87%卒中风险，可我翻遍他昨天的血常规、凝血四项、颈动脉超声报告，没一处超标。”最后上线的，是现在这个stroke-prediction-model：它不输出孤立概率，而是绑定临床检查项做条件推理——输入必须包含至少3项实测指标（如收缩压≥140mmHg + 空腹血糖≥7.0mmol/L + 颈动脉IMT≥0.9mm），缺一项就拒绝预测；输出不是“高/中/低风险”，而是“触发预警：需24小时内复查同型半胱氨酸+心电图”。它用 Jupyter Notebook 封装，但核心逻辑完全可导出为轻量 Python 模块，已嵌入某省基层医疗信息系统（HIS）的检验报告页脚。适合两类人：一是想快速验证临床规则与机器学习融合可行性的算法工程师；二是需要在无GPU、无运维支持的旧HIS环境里落地AI能力的医院信息科同事。它不追求SOTA，只解决一个具体问题：让预测结果能被医生一眼看懂、立刻执行。

2. 技术栈与工程结构：为什么选Jupyter Notebook做主载体，而不是Flask或FastAPI

提示：这不是“为了交互而交互”的Notebook，而是把开发-验证-交付流程压缩进单文件的工程选择。所有模型训练、特征工程、临床规则注入、接口封装都在同一个.ipynb里完成，最终导出的model_api.py仅依赖numpy和scikit-learn，零外部服务依赖。

2.1 核心依赖与版本锚点：轻量化落地的关键约束

项目严格锁定以下依赖组合，确保在Windows Server 2012 R2（某地市医院HIS服务器主流环境）上无需conda或docker即可运行：

# requirements.txt（全文仅5行） numpy==1.21.6 scikit-learn==1.0.2 pandas==1.3.5 joblib==1.1.0 jupyter==1.0.0

为什么是这些版本？scikit-learn 1.0.2是最后一个兼容joblib 1.1.0（序列化兼容性极强）且不强制要求threadpoolctl的版本；numpy 1.21.6能在Python 3.7–3.9全系稳定运行，避免某医院检验科电脑上因Python 3.10升级导致的ABI崩溃。我一般会先在目标环境跑python -c "import numpy; print(numpy.__version__)"，再反向匹配requirements——这步省掉，后面90%的“本地能跑线上报错”都源于此。

2.2 目录结构：Notebook即部署包，无隐藏配置文件

整个项目解压后只有4个文件，无子目录：

stroke-prediction-model/ ├── stroke_prediction_demo.ipynb # 主Notebook：含数据加载、训练、规则注入、API导出全流程 ├── data_sample.csv # 示例数据：1200例真实脱敏检验记录（含label） ├── model_api.py # 从Notebook导出的纯函数式API（无类、无状态） └── README.md # 仅两句话：①双击打开.ipynb ②修改data_sample.csv路径后按Shift+Enter全跑

注意：model_api.py不是手写的，而是由Notebook中%%writefile model_api.py单元格自动生成。这意味着你改了Notebook里的预测逻辑，重新运行该单元格，model_api.py就自动更新——没有“代码写在Notebook里，部署用另一个py文件”的割裂感。

2.3 模型架构设计：三层决策流，不是端到端黑匣子

模型本质是临床规则引擎 + 轻量梯度提升树（LightGBM）的混合体，结构如下：

关键设计点：Rule Layer拥有最高优先级。当rule_alert=True时，直接返回risk_class="high"，模型层输出被忽略。这保证了“指南明确要求干预”的病例永不被模型低估。

3. 核心模块拆解：从Notebook单元格到可部署API的完整链路

3.1 数据加载与硬规则校验：load_and_validate_data()函数

这是整个流程的入口，也是唯一允许用户修改的“业务入口点”。它不只读CSV，还强制执行临床合理性检查：

# 来自 stroke_prediction_demo.ipynb 的单元格 import pandas as pd import numpy as np def load_and_validate_data(filepath: str) -> pd.DataFrame: """ 加载数据并执行三级校验： 1. 列名存在性校验（必须含['sbp','fbg','imt','age','sex']） 2. 数值范围校验（如sbp必须在50-250mmHg之间） 3. 临床逻辑校验（如imt缺失时，fbg必须存在——因两者常联合检测） """ df = pd.read_csv(filepath) # 1. 列名校验 required_cols = ['sbp', 'fbg', 'imt', 'age', 'sex'] missing_cols = [c for c in required_cols if c not in df.columns] if missing_cols: raise ValueError(f"缺失必需列: {missing_cols}") # 2. 数值范围校验（示例：收缩压） if not ((df['sbp'] >= 50) & (df['sbp'] <= 250)).all(): invalid_sbp = df[~((df['sbp'] >= 50) & (df['sbp'] <= 250))] raise ValueError(f"收缩压超出合理范围(50-250mmHg)，共{len(invalid_sbp)}条异常记录") # 3. 临床逻辑校验：imt缺失时fbg必须存在 imt_missing = df['imt'].isna() fbg_missing_when_imt_missing = df[imt_missing & df['fbg'].isna()] if len(fbg_missing_when_imt_missing) > 0: raise ValueError("颈动脉IMT缺失时，空腹血糖(FBG)必须提供，共{}条冲突".format( len(fbg_missing_when_imt_missing))) return df # 使用示例（Notebook中直接调用） df_raw = load_and_validate_data("data_sample.csv") print(f"✅ 加载成功：{len(df_raw)}条记录，字段完整，数值合规")

逻辑说明：这个函数把“数据质量”和“临床逻辑”提前卡死在第一步。参数filepath必须是绝对路径或相对于Notebook的相对路径；校验失败时抛出带具体行数的ValueError，而非静默填充NaN——这是基层医院数据脏乱差场景下的血泪经验：宁可中断，不传错误。

3.2 硬规则引擎实现：apply_clinical_rules()的逐条映射

规则不是写在配置文件里，而是硬编码在函数中，每一条对应指南原文编号，方便审计：

def apply_clinical_rules(row: pd.Series) -> dict: """ 应用《中国缺血性卒中二级预防指南2022》核心条款 返回字典：{'alert': bool, 'reason': str, 'urgent_action': str} """ alert = False reason = "" urgent_action = "" # 条款3.2.1：高血压合并糖尿病患者，SBP≥140mmHg即触发 if row['sbp'] >= 140 and row['fbg'] >= 7.0: alert = True reason += "高血压合并糖尿病(SBP≥140且FBG≥7.0); " urgent_action += "24h内复查同型半胱氨酸+心电图; " # 条款4.1.3：颈动脉IMT≥0.9mm且年龄≥65岁，触发 if pd.notna(row['imt']) and row['imt'] >= 0.9 and row['age'] >= 65: alert = True reason += f"颈动脉IMT({row['imt']}mm)≥0.9mm且年龄({row['age']}岁)≥65; " urgent_action += "72h内安排颈动脉超声造影; " # 条款5.3.2：男性+吸烟史+LDL-C≥3.4mmol/L（本数据集暂无LDL-C，跳过） # （此处留空，体现规则可扩展性） return { "alert": alert, "reason": reason.rstrip("; "), "urgent_action": urgent_action.rstrip("; ") } # 在Notebook中批量应用 df_raw['rule_result'] = df_raw.apply(apply_clinical_rules, axis=1) df_raw['rule_alert'] = df_raw['rule_result'].apply(lambda x: x['alert']) print(f"⚠️ 硬规则触发预警：{df_raw['rule_alert'].sum()} / {len(df_raw)} 例")

参数说明：row是单行pd.Series，确保函数可被apply()向量化调用；reason字段用分号分隔，方便前端按;切分展示多条原因；urgent_action直接给出下一步操作指令，而非模糊描述。

3.3 特征工程与模型训练：build_features_and_train()的确定性流程

所有特征变换使用sklearn.preprocessing.StandardScaler的fit_transform，但固定随机种子并保存scaler对象，确保生产环境特征缩放一致：

from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split from lightgbm import LGBMClassifier import joblib def build_features_and_train(df: pd.DataFrame, random_state: int = 42): # 步骤1：构造特征矩阵X（12维） X = pd.DataFrame() X['sbp_normalized'] = (df['sbp'] - 120) / 20 # 均值120，标准差20 X['fbg_log'] = np.log1p(df['fbg']) # log1p处理偏态 X['imt_binary'] = (df['imt'] >= 0.9).astype(int) X['age_group_65plus'] = (df['age'] >= 65).astype(int) X['sex_male'] = (df['sex'] == 1).astype(int) # ... 其余7维（如BMI分段、既往卒中史等）省略 # 步骤2：标签y（二分类：1=1年内发生卒中，0=未发生） y = df['label'] # data_sample.csv中已含label列 # 步骤3：划分训练/测试集（固定random_state） X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=random_state, stratify=y ) # 步骤4：标准化（注意：只对X_train fit，X_test仅transform） scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) X_test_scaled = scaler.transform(X_test) # 关键！不重新fit # 步骤5：训练LightGBM（超参已调优，见Notebook历史记录） model = LGBMClassifier( n_estimators=50, max_depth=3, learning_rate=0.1, random_state=random_state, verbose=-1 ) model.fit(X_train_scaled, y_train) # 步骤6：保存scaler和model（供model_api.py加载） joblib.dump(scaler, "scaler.joblib") joblib.dump(model, "lgbm_model.joblib") return model, scaler, X_test_scaled, y_test # Notebook中调用 model, scaler, X_test, y_test = build_features_and_train(df_raw) print("✅ 特征构建与模型训练完成，scaler和model已保存")

逻辑说明：StandardScaler的fit_transform只在训练集上调用，测试集用transform——这是新手最易翻车的点；joblib.dump保存的对象，后续model_api.py通过joblib.load加载，路径硬编码为相对路径"scaler.joblib"，避免部署时路径错位。

4. 部署与运行：从Notebook一键生成API，零配置接入旧系统

4.1model_api.py的极简设计：无Web框架，纯函数调用

该文件由Notebook中以下单元格生成，全文仅127行，无任何类定义、无全局变量、无日志配置：

# Notebook中执行此单元格，生成model_api.py %%writefile model_api.py import numpy as np import pandas as pd import joblib # 加载训练好的组件（路径固定，不接受参数） scaler = joblib.load("scaler.joblib") model = joblib.load("lgbm_model.joblib") def predict_risk( sbp: float, fbg: float, imt: float, age: int, sex: int, # 其他6个特征参数（如bmi, ldl_c, smoking等）... ) -> dict: """ 主预测函数：输入原始检验值，返回结构化结果 """ # Step 1: 硬规则校验（复用Notebook中apply_clinical_rules逻辑） rule_alert, rule_reason, urgent_action = _apply_hard_rules(sbp, fbg, imt, age, sex) if rule_alert: return { "risk_class": "high", "risk_proba": 1.0, "rule_triggered": True, "rule_reason": rule_reason, "urgent_action": urgent_action, "model_used": False } # Step 2: 构建特征向量（复用Notebook中build_features逻辑） features = _build_feature_vector(sbp, fbg, imt, age, sex, ...) # Step 3: 标准化 + 预测 features_scaled = scaler.transform([features]) proba = model.predict_proba(features_scaled)[0][1] # P(class=1) risk_class = "high" if proba >= 0.6 else "medium" if proba >= 0.3 else "low" return { "risk_class": risk_class, "risk_proba": float(proba), "rule_triggered": False, "model_used": True } # 内部函数（_开头表示私有） def _apply_hard_rules(sbp, fbg, imt, age, sex): alert = False reason = "" action = "" if sbp >= 140 and fbg >= 7.0: alert = True reason = "高血压合并糖尿病" action = "24h内复查同型半胱氨酸+心电图" if not np.isnan(imt) and imt >= 0.9 and age >= 65: alert = True reason += "; " + "颈动脉IMT增厚伴高龄" action += "; " + "72h内安排颈动脉超声造影" return alert, reason.strip(), action.strip() def _build_feature_vector(sbp, fbg, imt, age, sex, ...): # 复制Notebook中build_features_and_train()的特征构造逻辑 # 注意：所有计算必须与Notebook中完全一致（如log1p、除法精度） return [sbp_norm, fbg_log, imt_bin, age_bin, sex_bin, ...] # 模块入口：直接运行时演示 if __name__ == "__main__": result = predict_risk(sbp=152, fbg=7.8, imt=1.1, age=72, sex=1) print(result)

关键点：predict_risk()函数签名与医院HIS系统调用习惯对齐——每个参数都是明确的检验项目名，而非抽象的feature_1,feature_2；返回字典字段名（如risk_class,urgent_action）可直接被HIS前端JSON解析渲染。

4.2 在旧HIS系统中集成：三行Python调用，无需重启服务

某地市医院信息科采用的方式：将model_api.py放入HIS服务器Python环境（Python 3.8.10），在检验报告生成脚本末尾插入：

# HIS系统原有报告生成脚本（report_generator.py）末尾追加 from model_api import predict_risk # 从数据库查出当前患者最新检验值 sbp = get_latest_value(patient_id, "收缩压") fbg = get_latest_value(patient_id, "空腹血糖") imt = get_latest_value(patient_id, "颈动脉IMT") age = get_patient_age(patient_id) sex = 1 if get_patient_sex(patient_id) == "男" else 0 # 调用预测 result = predict_risk(sbp=sbp, fbg=fbg, imt=imt, age=age, sex=sex) # 将result['urgent_action']写入报告页脚 report_footer += f"【AI预警】{result['urgent_action']}"

注意：get_latest_value()是HIS系统原有函数，无需改动；predict_risk()调用耗时<15ms（实测i5-4590 CPU），不影响报告生成速度。

4.3 常见问题排查：5条真实踩坑记录

现象1：HIS服务器上运行model_api.py报错ModuleNotFoundError: No module named 'lightgbm'

原因：HIS服务器Python环境未安装LightGBM，且因网络策略无法pip install。
解决：在开发机上用pip download lightgbm==3.3.5 --no-deps --platform manylinux2014_x86_64 --python-version 38 --only-binary=:all:下载.whl文件，手动拷贝至HIS服务器，执行pip install lightgbm-3.3.5-py3-none-manylinux2014_x86_64.whl。注意平台和Python版本必须严格匹配。

现象2：预测结果risk_proba全为0.0或1.0，且与Notebook中结果不一致

原因：model_api.py中scaler.transform([features])的输入是单样本列表[features]，若误写为scaler.transform(features)（无方括号），则scaler会尝试对1D数组按列标准化，导致维度错乱。
解决：检查_build_feature_vector()返回的是1Dnp.ndarray（shape=(12,)），传入scaler前必须包装成2D[[f1,f2,...,f12]]。

现象3：HIS调用时偶尔返回ValueError: Input contains NaN

原因：HIS数据库中某些检验值为NULL，get_latest_value()返回None，传入predict_risk()后在_apply_hard_rules()中参与比较（如None >= 140）不报错，但进入_build_feature_vector()后np.log1p(None)产生NaN。
解决：在predict_risk()开头增加参数校验：

if any(np.isnan([sbp, fbg, imt, age, sex])): return {"error": "输入含空值，请检查检验数据完整性"}

现象4：joblib.load("scaler.joblib")报错FileNotFoundError

原因：HIS系统调用model_api.py时工作目录不是项目根目录，相对路径"scaler.joblib"解析失败。
解决：在model_api.py开头添加路径修正：

import os BASE_DIR = os.path.dirname(os.path.abspath(__file__)) scaler = joblib.load(os.path.join(BASE_DIR, "scaler.joblib")) model = joblib.load(os.path.join(BASE_DIR, "lgbm_model.joblib"))

现象5：模型预测结果与医生经验严重不符，如年轻健康者被判high风险

原因：特征工程中sbp_normalized = (sbp - 120) / 20的均值120来自训练集统计，但HIS中年轻患者SBP普遍偏低（如110），导致(110-120)/20 = -0.5进入模型，而模型在训练时几乎没见过负值SBP特征。
解决：在_build_feature_vector()中增加截断：

sbp_norm = max(-2.0, min(3.0, (sbp - 120) / 20)) # 限定在[-2,3]区间

这比用训练集min/max更鲁棒——临床值域外的输入，应被主动钳位，而非让模型外推。

5. 二次开发实战：如何安全替换模型、注入新规则、适配新检验项目

5.1 替换模型：保留规则层，只更新模型层（3步完成）

当你有新数据或想试用XGBoost时，无需改动model_api.py，只需在Notebook中操作：

1. 替换训练代码：在build_features_and_train()单元格中，将LGBMClassifier替换为XGBClassifier，保持特征维度和标签不变；
2. 重新运行训练单元格：joblib.dump会覆盖lgbm_model.joblib为xgb_model.joblib；
3. 修改model_api.py加载逻辑：找到model = joblib.load("lgbm_model.joblib")行，改为model = joblib.load("xgb_model.joblib")。

关键原则：模型文件名是唯一耦合点，规则层和特征层完全隔离。这样即使模型崩了，硬规则仍可用。

5.2 注入新临床规则：修改两个函数，无需重训模型

假设新指南增加“房颤患者CHADS₂-VASc评分≥2即预警”，你需要：

1. 在_apply_hard_rules()中添加逻辑：

# 新增参数 afib: bool (是否房颤), chads2_vasc: int if afib and chads2_vasc >= 2: alert = True reason += "; " + "房颤伴CHADS₂-VASc≥2" action += "; " + "立即启动抗凝评估"

2. 在predict_risk()函数签名中增加参数：

def predict_risk( sbp: float, fbg: float, imt: float, age: int, sex: int, afib: bool, chads2_vasc: int # ← 新增 ) -> dict:

3. 同步更新model_api.py的if __name__ == "__main__":演示代码，确保调用示例完整。

注意：新增参数必须有默认值（如afib: bool = False），否则旧HIS系统不传参会报错。这是向前兼容的底线。

5.3 适配新检验项目：扩展特征向量（5处同步修改）

例如医院新上“血清神经丝轻链（NfL）”检测，想纳入模型：

血泪教训：我在某次升级中只改了Notebook和model_api.py，忘了通知信息科在HIS中补NfL字段提取，结果所有预测返回error: input contains NaN。从那以后我每次新增字段，都强制走一遍“Notebook→model_api.py→HIS伪代码注释→信息科确认邮件”四步闭环。

5.4 验证新版本正确性：用data_sample.csv做回归测试

项目自带的data_sample.csv不是训练集，而是黄金测试集——它包含10条典型case（如“SBP=160+FBG=8.2→rule_alert=True”，“SBP=110+FBG=5.0→model_proba=0.12”）。在修改任何逻辑后，必须运行：

# 在Notebook末尾新增单元格 test_cases = pd.read_csv("data_sample.csv").head(10) results = [] for _, row in test_cases.iterrows(): r = predict_risk( sbp=row['sbp'], fbg=row['fbg'], imt=row['imt'], age=row['age'], sex=row['sex'] # 其他参数... ) results.append(r) # 人工核对results是否符合预期（如第1条必须rule_alert=True） for i, r in enumerate(results): print(f"Case {i+1}: {r['risk_class']} (rule:{r.get('rule_triggered',False)})")

这不是自动化测试，但足够快、足够准。真正的自动化测试（pytest）可以后续加，但在基层医院场景下，一份清晰的手动验证清单，比100行pytest更可靠——因为信息科同事能看懂、能执行、能签字确认。

希望帮到你。

本文还有配套的精品资源，点击获取