- Authors
- Name
- Youngju Kim
- @fjvbn20031
概述
AI 正在医疗健康与金融这两大产业中引发前所未有的变革。从医生用于影像判读的深度学习模型，到每秒处理数千笔交易的算法交易系统，AI 已经成为这些领域的核心基础设施。本指南将结合实际代码，深入探讨这两个领域的主要应用案例。
Part 1: 医疗健康 AI
1. 医学影像 AI
医学影像分析是深度学习展现最大影响力的领域之一。CheXNet（斯坦福大学）模型在胸部 X 光片上实现了达到放射科专家水平的肺炎诊断性能，此后成为众多医学影像 AI 研究的基础。
DICOM 图像处理
这是用 Python 处理医学影像标准格式 DICOM 的基本流水线。
import pydicom
import numpy as np
from torchvision import transforms
import torch
# 加载 DICOM 图像
ds = pydicom.dcmread('chest_xray.dcm')
img_array = ds.pixel_array.astype(np.float32)
# 归一化（HU 转 Windowing）
img_normalized = (img_array - img_array.min()) / (img_array.max() - img_array.min())
# 用于模型预测的预处理
transform = transforms.Compose([
transforms.ToPILImage(),
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize([0.485], [0.229])
])
tensor = transform(img_normalized).unsqueeze(0)
使用 U-Net 进行医学影像分割
为了在 MRI 或 CT 中自动分割肿瘤、器官等感兴趣区域，U-Net 架构被广泛使用。
import torch
import torch.nn as nn
class DoubleConv(nn.Module):
def __init__(self, in_ch, out_ch):
super().__init__()
self.conv = nn.Sequential(
nn.Conv2d(in_ch, out_ch, 3, padding=1),
nn.BatchNorm2d(out_ch),
nn.ReLU(inplace=True),
nn.Conv2d(out_ch, out_ch, 3, padding=1),
nn.BatchNorm2d(out_ch),
nn.ReLU(inplace=True)
)
def forward(self, x):
return self.conv(x)
class UNet(nn.Module):
def __init__(self, in_channels=1, num_classes=2):
super().__init__()
self.enc1 = DoubleConv(in_channels, 64)
self.enc2 = DoubleConv(64, 128)
self.pool = nn.MaxPool2d(2)
self.bottleneck = DoubleConv(128, 256)
self.up1 = nn.ConvTranspose2d(256, 128, 2, stride=2)
self.dec1 = DoubleConv(256, 128)
self.up2 = nn.ConvTranspose2d(128, 64, 2, stride=2)
self.dec2 = DoubleConv(128, 64)
self.out_conv = nn.Conv2d(64, num_classes, 1)
def forward(self, x):
e1 = self.enc1(x)
e2 = self.enc2(self.pool(e1))
b = self.bottleneck(self.pool(e2))
d1 = self.dec1(torch.cat([self.up1(b), e2], dim=1))
d2 = self.dec2(torch.cat([self.up2(d1), e1], dim=1))
return self.out_conv(d2)
评估指标：在医学分割中，Dice Score 与 IoU（Intersection over Union，交并比）是主要指标。Dice Score 表示两个集合的重叠比例，在背景-前景不均衡的问题中比 Accuracy 更有用。
2. 临床 NLP
电子健康记录（EHR）中存在大量非结构化文本数据。NLP 模型从这些数据中提取具有临床意义的信息。
医疗命名实体识别（NER）
from transformers import pipeline, AutoTokenizer, AutoModelForTokenClassification
model_name = "allenai/scibert_scivocab_uncased"
tokenizer = AutoTokenizer.from_pretrained(model_name)
# 基于 BioBERT 的 NER 流水线
ner_pipeline = pipeline(
"ner",
model="dmis-lab/biobert-v1.1",
tokenizer="dmis-lab/biobert-v1.1",
aggregation_strategy="simple"
)
clinical_text = """
Patient presents with acute myocardial infarction.
Administered aspirin 325mg and clopidogrel 600mg.
ECG shows ST elevation in leads V1-V4.
"""
entities = ner_pipeline(clinical_text)
for ent in entities:
print(f"Entity: {ent['word']}, Label: {ent['entity_group']}, Score: {ent['score']:.3f}")
临床文本摘要（应用 Med-PaLM、BioMedLM）
PhysioNet 的 MIMIC-III 数据集是临床 NLP 研究的标准基准。它包含出院小结、放射学报告、护理记录等多种临床文本，已获得访问权限的研究者可将其用于模型训练。
3. 新药研发 AI
新药研发平均需要耗时 10 到 15 年、花费数十亿美元。AI 正在加速这一流程中的多个环节。
分子特性预测（应用 RDKit）
from rdkit import Chem
from rdkit.Chem import Descriptors
import pandas as pd
def compute_molecular_features(smiles_list):
features = []
for smiles in smiles_list:
mol = Chem.MolFromSmiles(smiles)
if mol:
features.append({
'MW': Descriptors.MolWt(mol),
'LogP': Descriptors.MolLogP(mol),
'HBD': Descriptors.NumHDonors(mol),
'HBA': Descriptors.NumHAcceptors(mol),
'TPSA': Descriptors.TPSA(mol)
})
return pd.DataFrame(features)
def lipinski_filter(df):
"""利宾斯基 5 规则：口服可给药性过滤器"""
return df[
(df['MW'] <= 500) &
(df['LogP'] <= 5) &
(df['HBD'] <= 5) &
(df['HBA'] <= 10)
]
# 示例 SMILES 列表
smiles_list = [
'CC(=O)Oc1ccccc1C(=O)O', # 阿司匹林
'CC12CCC3C(C1CCC2O)CCC4=CC(=O)CCC34C', # 睾酮
]
df = compute_molecular_features(smiles_list)
drug_candidates = lipinski_filter(df)
AlphaFold2 与蛋白质结构预测
DeepMind 的 AlphaFold2 是一个能从蛋白质序列预测 3D 结构的突破性模型。它实质上解决了困扰学界数十年的蛋白质折叠问题（Protein Folding Problem），改变了新药研发的范式。
# ColabFold（AlphaFold2 接口）使用示例
# pip install colabfold
from colabfold.batch import get_queries, run
# 输入蛋白质序列（FASTA 格式）
queries = [("target_protein", "MKTIIALSYIFCLVFA")]
results = run(
queries=queries,
result_dir="./alphafold_results",
use_templates=False,
num_recycles=3,
model_type="auto"
)
使用 GNN 生成分子
图神经网络（GNN）将分子表示为原子（节点）与化学键（边）的图结构，用于生成新的候选药物。
import torch
from torch_geometric.data import Data
from torch_geometric.nn import GCNConv, global_mean_pool
class MoleculeGNN(torch.nn.Module):
def __init__(self, num_features, hidden_dim, num_classes):
super().__init__()
self.conv1 = GCNConv(num_features, hidden_dim)
self.conv2 = GCNConv(hidden_dim, hidden_dim)
self.classifier = torch.nn.Linear(hidden_dim, num_classes)
self.relu = torch.nn.ReLU()
def forward(self, data):
x, edge_index, batch = data.x, data.edge_index, data.batch
x = self.relu(self.conv1(x, edge_index))
x = self.relu(self.conv2(x, edge_index))
x = global_mean_pool(x, batch)
return self.classifier(x)
4. 可穿戴数据与 AI
从智能手表和可穿戴设备中收集到的生理信号数据，正在开启实时健康监测的全新范式。
ECG 心律失常分类（1D CNN）
import torch
import torch.nn as nn
class ECGClassifier(nn.Module):
"""1D CNN for ECG arrhythmia classification"""
def __init__(self, num_classes=5):
super().__init__()
self.features = nn.Sequential(
nn.Conv1d(1, 32, kernel_size=11, padding=5),
nn.BatchNorm1d(32),
nn.ReLU(),
nn.MaxPool1d(2),
nn.Conv1d(32, 64, kernel_size=7, padding=3),
nn.BatchNorm1d(64),
nn.ReLU(),
nn.MaxPool1d(2),
nn.Conv1d(64, 128, kernel_size=5, padding=2),
nn.BatchNorm1d(128),
nn.ReLU(),
nn.AdaptiveAvgPool1d(32)
)
self.classifier = nn.Sequential(
nn.Linear(128 * 32, 256),
nn.ReLU(),
nn.Dropout(0.5),
nn.Linear(256, num_classes)
)
def forward(self, x):
x = self.features(x)
x = x.view(x.size(0), -1)
return self.classifier(x)
# 应用 PhysioNet MIT-BIH 心律失常数据库
# https://physionet.org/content/mitdb/1.0.0/
主要 ECG 类别：对 Normal（N）、Atrial Fibrillation（A）、Other（O）、Noisy（~）等多种心律失常模式进行分类。
Part 2: 金融 AI
5. 股价预测模型
股价预测是金融 AI 中被研究得最广泛的领域，但同时也是最困难的课题。根据有效市场假说（EMH），公开信息已经反映在价格之中，因此仅靠价格数据本身，极难持续获得超额收益。
运用技术指标进行特征工程
import yfinance as yf
import pandas as pd
import numpy as np
# 使用 yfinance 下载股价数据
# 参考：https://pypi.org/project/yfinance/
ticker = yf.Ticker("AAPL")
df = ticker.history(period="5y")
def compute_rsi(prices, window=14):
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window).mean()
rs = gain / loss
return 100 - (100 / (1 + rs))
def add_technical_indicators(df):
df['SMA20'] = df['Close'].rolling(20).mean()
df['SMA50'] = df['Close'].rolling(50).mean()
df['EMA12'] = df['Close'].ewm(span=12).mean()
df['EMA26'] = df['Close'].ewm(span=26).mean()
df['MACD'] = df['EMA12'] - df['EMA26']
df['Signal'] = df['MACD'].ewm(span=9).mean()
df['RSI'] = compute_rsi(df['Close'], 14)
df['BB_upper'] = df['SMA20'] + 2 * df['Close'].rolling(20).std()
df['BB_lower'] = df['SMA20'] - 2 * df['Close'].rolling(20).std()
df['Volatility'] = df['Close'].pct_change().rolling(20).std()
df['Volume_MA'] = df['Volume'].rolling(20).mean()
return df.dropna()
df = add_technical_indicators(df)
基于 LSTM 的时间序列预测
import torch
import torch.nn as nn
from sklearn.preprocessing import MinMaxScaler
class StockLSTM(nn.Module):
def __init__(self, input_size, hidden_size=128, num_layers=2, dropout=0.2):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout
)
self.fc = nn.Sequential(
nn.Linear(hidden_size, 64),
nn.ReLU(),
nn.Linear(64, 1)
)
def forward(self, x):
out, _ = self.lstm(x)
return self.fc(out[:, -1, :])
# Walk-Forward 交叉验证（防止 Look-Ahead Bias）
def walk_forward_split(df, train_size=0.7, val_size=0.15):
n = len(df)
train_end = int(n * train_size)
val_end = int(n * (train_size + val_size))
return df[:train_end], df[train_end:val_end], df[val_end:]
重要：切勿对时间序列数据使用随机切分。这会导致未来数据混入训练集，产生 look-ahead bias，使回测结果看起来比实际情况好得多。
6. 算法交易
使用 Backtrader 实现移动平均交叉策略
import backtrader as bt
# 参考：https://www.backtrader.com/docu/
class MACrossStrategy(bt.Strategy):
params = (('fast', 10), ('slow', 30),)
def __init__(self):
self.sma_fast = bt.indicators.SMA(period=self.p.fast)
self.sma_slow = bt.indicators.SMA(period=self.p.slow)
self.crossover = bt.indicators.CrossOver(self.sma_fast, self.sma_slow)
def next(self):
if not self.position:
if self.crossover > 0:
self.buy(size=100)
elif self.crossover < 0:
self.sell(size=100)
# 执行回测
cerebro = bt.Cerebro()
cerebro.addstrategy(MACrossStrategy)
cerebro.broker.setcash(100000.0)
cerebro.broker.setcommission(commission=0.001)
results = cerebro.run()
print(f"Final Portfolio Value: {cerebro.broker.getvalue():.2f}")
投资组合优化（Markowitz 均值方差）
import numpy as np
from scipy.optimize import minimize
def portfolio_stats(weights, returns):
port_return = np.sum(returns.mean() * weights) * 252
port_vol = np.sqrt(weights @ returns.cov() @ weights * 252)
sharpe = port_return / port_vol
return port_return, port_vol, sharpe
def min_variance(returns):
n = returns.shape[1]
constraints = {'type': 'eq', 'fun': lambda w: np.sum(w) - 1}
bounds = [(0, 1)] * n
result = minimize(
lambda w: portfolio_stats(w, returns)[1],
x0=np.ones(n) / n,
bounds=bounds,
constraints=constraints
)
return result.x
基于强化学习的交易（PPO）
import gym
import numpy as np
from stable_baselines3 import PPO
class TradingEnv(gym.Env):
"""简化版股票交易环境"""
def __init__(self, df, initial_balance=10000):
super().__init__()
self.df = df
self.initial_balance = initial_balance
self.action_space = gym.spaces.Discrete(3) # 0: 持有, 1: 买入, 2: 卖出
self.observation_space = gym.spaces.Box(
low=-np.inf, high=np.inf, shape=(10,), dtype=np.float32
)
self.reset()
def reset(self):
self.balance = self.initial_balance
self.shares = 0
self.current_step = 0
return self._get_obs()
def _get_obs(self):
row = self.df.iloc[self.current_step]
return np.array([
row['Close'], row['SMA20'], row['SMA50'],
row['RSI'], row['MACD'], row['Volatility'],
self.balance, self.shares,
row['Volume'], row['BB_upper'] - row['BB_lower']
], dtype=np.float32)
def step(self, action):
price = self.df.iloc[self.current_step]['Close']
reward = 0
if action == 1 and self.balance >= price:
self.shares += 1
self.balance -= price
elif action == 2 and self.shares > 0:
self.shares -= 1
self.balance += price
reward = price - self.df.iloc[max(0, self.current_step - 1)]['Close']
self.current_step += 1
done = self.current_step >= len(self.df) - 1
return self._get_obs(), reward, done, {}
7. 欺诈检测
信用卡欺诈检测面临极端类别不平衡（正常交易 99.9% vs 欺诈交易 0.1%）这一特殊挑战。
SMOTE 与 Isolation Forest
from sklearn.ensemble import IsolationForest
from imblearn.over_sampling import SMOTE
from sklearn.metrics import classification_report, roc_auc_score
import numpy as np
# 对不平衡数据进行过采样
smote = SMOTE(sampling_strategy=0.1, random_state=42)
X_resampled, y_resampled = smote.fit_resample(X_train, y_train)
# 无监督异常检测
iso_forest = IsolationForest(
n_estimators=200,
contamination=0.001,
random_state=42
)
iso_forest.fit(X_train)
anomaly_scores = iso_forest.decision_function(X_test)
# 用 SHAP 解释模型
import shap
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test[:100])
shap.summary_plot(shap_values, X_test[:100], feature_names=feature_names)
基于自编码器的异常检测
import torch
import torch.nn as nn
class FraudAutoencoder(nn.Module):
def __init__(self, input_dim, encoding_dim=16):
super().__init__()
self.encoder = nn.Sequential(
nn.Linear(input_dim, 64),
nn.ReLU(),
nn.Linear(64, 32),
nn.ReLU(),
nn.Linear(32, encoding_dim)
)
self.decoder = nn.Sequential(
nn.Linear(encoding_dim, 32),
nn.ReLU(),
nn.Linear(32, 64),
nn.ReLU(),
nn.Linear(64, input_dim)
)
def forward(self, x):
encoded = self.encoder(x)
decoded = self.decoder(encoded)
return decoded
# 重构误差较大则判定为欺诈
def detect_fraud(model, x, threshold=0.5):
reconstructed = model(x)
reconstruction_error = torch.mean((x - reconstructed) ** 2, dim=1)
return reconstruction_error > threshold
8. 信用风险模型
在巴塞尔 III 监管框架下，金融机构负有对信用风险进行量化的义务。预期损失（Expected Loss）计算方式如下。
- EL = PD x LGD x EAD
- PD（Probability of Default）：违约概率
- LGD（Loss Given Default）：违约损失率
- EAD（Exposure at Default）：违约风险敞口
from sklearn.linear_model import LogisticRegression
from sklearn.calibration import CalibratedClassifierCV
import lightgbm as lgb
# PD 模型（逻辑回归）
pd_model = LogisticRegression(class_weight='balanced', max_iter=1000)
pd_model_calibrated = CalibratedClassifierCV(pd_model, cv=5)
pd_model_calibrated.fit(X_train, y_train)
pd_scores = pd_model_calibrated.predict_proba(X_test)[:, 1]
# 用 LightGBM 实现评分卡
lgb_model = lgb.LGBMClassifier(
n_estimators=500,
learning_rate=0.05,
num_leaves=31,
class_weight='balanced'
)
lgb_model.fit(X_train, y_train, eval_set=[(X_val, y_val)])
# 计算预期损失
def expected_loss(pd, lgd, ead):
return pd * lgd * ead
# Survival Analysis for Time-to-Default
from lifelines import CoxPHFitter
cph = CoxPHFitter()
cph.fit(df_survival, duration_col='time_to_default', event_col='defaulted')
9. 金融 NLP
新闻情绪分析与股价相关性
from transformers import pipeline
import pandas as pd
# 应用 FinanceBERT（金融领域专用 BERT）
sentiment_pipeline = pipeline(
"sentiment-analysis",
model="ProsusAI/finbert"
)
def analyze_news_sentiment(headlines):
results = sentiment_pipeline(headlines)
return pd.DataFrame([
{'headline': h, 'label': r['label'], 'score': r['score']}
for h, r in zip(headlines, results)
])
headlines = [
"Apple reports record quarterly earnings",
"Federal Reserve signals rate hike ahead",
"Tech sector faces regulatory headwinds"
]
df_sentiment = analyze_news_sentiment(headlines)
print(df_sentiment)
解析 EDGAR 财务报表
import requests
from bs4 import BeautifulSoup
import re
def fetch_10k_filing(cik, accession_number):
"""从 SEC EDGAR 获取 10-K 年度报告"""
base_url = "https://www.sec.gov/Archives/edgar"
url = f"{base_url}/{cik}/{accession_number}/10k.htm"
response = requests.get(url, headers={'User-Agent': 'research@example.com'})
soup = BeautifulSoup(response.text, 'html.parser')
# 提取特定章节（例如 Risk Factors）
risk_section = soup.find(text=re.compile("Risk Factors", re.IGNORECASE))
return risk_section.parent.get_text() if risk_section else ""
医疗健康与金融 AI 核心注意事项
医疗健康
- FDA 监管：在美国，医疗 AI 软件受 FDA 的 Software as a Medical Device（SaMD）框架监管。
- HIPAA 合规：处理患者数据时必须进行去标识化（De-identification）并采取安全措施。
- 偏差（Bias）问题：若特定种族或性别的数据不足，AI 模型对该群体的表现可能会下降。
- 临床验证：即使模型性能优异，在真实临床环境中的验证（Clinical Trial）仍是必不可少的。
金融
- Look-Ahead Bias：训练时间序列模型时，必须彻底杜绝未来信息的泄漏。
- Overfitting：金融数据噪声较多，复杂模型在实盘中反而可能表现更差。
- 市场机制变化：过去表现良好的策略，可能因市场环境变化而突然失效。
- 交易成本：回测中必须计入滑点、手续费和市场冲击成本。
参考资料
- PhysioNet Clinical Data：https://physionet.org/
- AlphaFold2（DeepMind）：https://alphafold.ebi.ac.uk/
- yfinance 官方文档：https://pypi.org/project/yfinance/
- Backtrader 官方文档：https://www.backtrader.com/docu/
- MIMIC-III Clinical Database：https://physionet.org/content/mimiciii/
- SEC EDGAR：https://www.sec.gov/edgar/
小测验
Q1. 金融 AI 回测中的 look-ahead bias 是什么，为什么危险？
答案：指在回测中，把该时间点原本无法得知的未来数据用于模型训练或信号生成的错误。
解释：例如，如果用今天的收盘价数据来生成今天的交易信号，这在现实中是不可能执行的交易。可以通过 Walk-Forward 验证以及考虑时间顺序的交叉验证来防止这一问题。存在这种偏差的回测，会让人误以为发现了实际上并不存在的 alpha，在实际投资时可能导致重大损失。
Q2. 医学影像 AI 中为什么用 Dice Score 而不是 Accuracy？
答案：因为在医学影像分割中，背景像素的数量远远超过感兴趣区域，类别不平衡极其严重。
解释：例如，在脑部 MRI 中，如果肿瘤只占全部像素的 1%，即使把所有像素都预测为背景，Accuracy 依然会达到 99%。Dice Score 的计算方式是 2 倍 TP 除以（2 倍 TP 加 FP 加 FN），能更准确地反映少数类别的预测性能。
Q3. 在新药研发中，利宾斯基 5 规则（Ro5）的使用目的是什么？
答案：这是一套用于在早期阶段筛选具备口服给药可能性的候选药物药代动力学特性的经验法则。
解释：分子量在 500 以下、LogP 在 5 以下、氢键供体在 5 个以下、氢键受体在 10 个以下的分子，在口服给药时具有较高的吸收和渗透性可能性。通过早期筛选，可以快速过滤数百万个候选分子，从而降低研发成本。
Q4. 在欺诈检测中使用 SMOTE 时需要注意什么？
答案：SMOTE 必须只应用于训练数据，绝不能应用于验证集或测试集。
解释：如果对测试集应用 SMOTE，会混入合成数据，从而高估实际性能。此外，SMOTE 生成的合成样本可能无法完全反映真实的欺诈模式，因此使用 Precision-Recall AUC 或 F1 Score 评估，比使用 Accuracy 更为合适。
Q5. AlphaFold2 对新药研发具有革命性意义的原因是什么？
答案：因为它仅凭蛋白质的氨基酸序列，就能以原子级精度预测其 3D 结构，把过去需要数年才能完成的蛋白质结构解析过程，缩短到了数分钟以内。
解释：在新药研发中，靶点蛋白质的 3D 结构对于确定药物结合位点（binding site）至关重要。在 AlphaFold2 出现之前，需要依靠 X 射线晶体学、冷冻电镜（cryo-EM）等实验方法。DeepMind 目前以公开数据库的形式，提供超过 2 亿个蛋白质结构预测结果。