本章目标：把 SAS 的
%MACRO思维迁移到 Python 函数； 学会用模块组织代码；学会用异常处理替代 SAS 的"日志 + 继续跑"。
4.1 控制流：与 DATA 步逐行逻辑几乎一一对应
if / elif / else
SAS
/* SAS DATA 步 */
data want;
set have;
if age < 65 then agegr = '<65';
else if age <= 80 then agegr = '65-80';
else agegr = '>80';
run;
Python
# Python（单值场景）
def age_group(age):
if age < 65:
return "<65"
elif age <= 80: # ← SAS 的 else if 在 Python 里是 elif
return "65-80"
else:
return ">80"
# 但处理整列数据时，用向量化（第 06 章）：
import numpy as np
df["AGEGR1"] = np.select(
[df["AGE"] < 65, df["AGE"] <= 80],
["<65", "65-80"],
default=">80",
)
💡 原则：
if/elif/else用于控制程序流程（配置判断、错误处理）； 数据的分组派生一律用向量化（np.where/np.select/pd.cut）。 这是"Python 思维"和"SAS 思维"的分水岭。
for 循环
SAS 里你几乎不写循环（DATA 步自动逐行，%DO 只用于宏生成代码）。
Python 里循环主要用于对一组"东西"做同一件事（文件、参数组合、数据集）。
SAS
/* SAS：%DO 循环通常用来生成代码 */
%macro loop;
%do i = 1 %to 3;
proc print data=ds&i.; run;
%end;
%mend;
Python
# Python：循环遍历"真实的对象"
for domain in ["dm", "ae", "cm", "ex"]:
print(f"处理 {domain}.xpt ...")
# 遍历带序号（对应 SAS 的自动宏变量 &i）
for i, domain in enumerate(["dm", "ae", "cm"], start=1):
print(f"{i}. {domain}")
# 遍历字典
var_labels = {"AGE": "Age", "SEX": "Sex"}
for name, label in var_labels.items():
print(f"{name:<10} {label}")
# range 循环
for i in range(3): # 0, 1, 2
print(i)
for i in range(1, 4): # 1, 2, 3
print(i)
while 循环
Python
# 用得不多的场景：分页抓取 API、重试
attempts = 0
while attempts < 3:
attempts += 1
print(f"第 {attempts} 次尝试")
break / continue
Python
for domain in ["dm", "ae", "error", "cm"]:
if domain == "error":
print("遇到错误域，停止")
break # 跳出整个循环
if domain == "ae":
continue # 跳过本次，继续下一次
print(f"处理 {domain}")
⚠️ Python 的
for有else子句（循环没被 break 时执行）—— 这是 SAS 里没有的概念，偶尔很好用，但不建议新手使用，容易误读。
4.2 函数：%MACRO 的替代品
这是最重要的迁移点。SAS 的宏本质是文本替换， Python 的函数是真正的参数传递 + 返回值——更安全、更易调试。
从宏到函数的翻译
SAS
/* SAS：一个生成人口学表的宏 */
%macro demo_tbl(in=, out=, trtvar=);
proc means data=&in. n mean std;
class &trtvar.;
var AGE;
run;
%mend demo_tbl;
%demo_tbl(in=adsl, out=t1, trtvar=TRT01P);
Python
# Python：一个生成人口学表的函数
def demo_table(df, trtvar="TRT01P", agevar="AGE"):
"""按治疗组汇总年龄的 N / Mean / SD。
Parameters
----------
df : DataFrame 分析数据集（如 ADSL）
trtvar : str 治疗组变量名
agevar : str 年龄变量名
Returns
-------
DataFrame 行为治疗组，列为 N / Mean / SD
"""
return (
df.groupby(trtvar)[agevar]
.agg(N="count", Mean="mean", SD="std")
.round(2)
)
t1 = demo_table(adsl, trtvar="TRT01P")
关键差异：
| SAS 宏 | Python 函数 | |
|---|---|---|
| 本质 | 文本替换 | 值传递 + 独立作用域 |
| 参数 | 宏变量（巨型全局变量） | 局部变量，不污染外部 |
| 返回 | 靠宏变量或输出数据集 | return 明确返回值 |
| 调试 | 很难（宏展开难读） | 容易（可打日志、可单元测试） |
| 复用 | %include + 调用宏 |
import 模块 + 调用函数 |
🔥 一个必须养成的习惯：写函数时加 docstring（三引号内的说明）。 这是 Python 的"程序头注释"，但更好——它会被 IDE 自动提示， 也是 QC 评审时的关键证据。
默认参数与关键字参数
Python
def summarize(df, var, by=None, decimals=2, na_rm=True):
"""通用描述统计。"""
if by:
res = df.groupby(by)[var].agg(["count", "mean", "std"])
else:
res = df[var].agg(["count", "mean", "std"])
return res.round(decimals)
summarize(adsl, "AGE") # 位置参数
summarize(adsl, var="AGE", by="TRT01P") # 关键字参数（推荐，可读性好）
summarize(adsl, "AGE", by="TRT01P", decimals=1) # 混合
💡 临床编程建议：调用时尽量用关键字参数（
by="TRT01P"）。 因为临床程序讲究可读性与可 QC 性，而关键字参数让每个参数的含义一目了然， 减少"参数顺序搞错"这类低级错误。
返回多个值
Python
def age_stats(df, var="AGE"):
s = df[var].dropna()
return len(s), s.mean(), s.std() # 返回 tuple
n, mean, sd = age_stats(adsl)
print(f"N={n}, Mean={mean:.1f}, SD={sd:.2f}")
args 与 *kwargs（了解即可）
Python
def make_table(title, *columns, **options):
print(f"标题: {title}")
print(f"列: {columns}")
print(f"选项: {options}")
make_table("Table 1", "TRT01P", "AGE", "SEX", font_size=10, page=1)
4.3 模块：把自己的代码拆成文件
SAS 用 %INCLUDE 复用代码。Python 用 import，但更严格、更清晰。
目录结构约定
文本
clinical-python-roadmap/
├── cases/
│ ├── case02_人口学表.py
│ └── case03_不良事件汇总表.py
├── clinic/ ← 你自己的工具包
│ ├── __init__.py ← 空文件，标记这是一个"包"
│ ├── io.py ← 读数据
│ ├── format.py ← 数字格式化、报表布局
│ └── qc.py ← QC 检查工具
写模块与导入
Python
# clinic/io.py
"""临床数据读取工具。"""
from pathlib import Path
import pandas as pd
def read_xpt(path, encoding="utf-8"):
"""读取 SAS XPORT (.xpt) 文件，返回 DataFrame。"""
return pd.read_sas(Path(path), format="xport", encoding=encoding)
def read_any(path):
"""按扩展名自动选择读取方式。"""
p = Path(path)
if p.suffix.lower() == ".xpt":
return read_xpt(p)
if p.suffix.lower() == ".csv":
return pd.read_csv(p)
if p.suffix.lower() == ".xlsx":
return pd.read_excel(p)
raise ValueError(f"不支持的文件类型: {p.suffix}")
Python
# cases/case02_人口学表.py
from clinic.io import read_any # ← 类似 %include clinic/io.sas
adsl = read_any("data/samples/adsl.csv")
三种导入方式：
Python
import pandas as pd # 推荐：导入模块，用 pd.xxx（最清晰）
from pathlib import Path # 导入具体对象
from clinic.io import read_xpt # 从自己的包导入
# 不要这样（会污染命名空间，且无法看出函数来自哪里）：
from pandas import *
💡 别名约定（行业惯例，照抄即可）：
import pandas as pd、import numpy as np、import matplotlib.pyplot as plt。
标准库：不用安装就有的"内置 SAS 函数库"
| 需求 | 模块 | SAS 对应 |
|---|---|---|
| 路径与文件操作 | pathlib |
LIBNAME + %sysfunc(pathname()) |
| 目录遍历 / 批处理 | pathlib.glob、os |
%sysfunc(dread())、PROC DATASETS |
| 日期时间 | datetime |
SAS 日期函数（today()、intck） |
| 正则 | re |
PRX* 函数族 |
| JSON / CSV | json、csv |
PROC JSON、PROC IMPORT |
| 随机数 | random |
RAND() |
| 日志 | logging |
SAS 日志机制 |
| 命令行参数 | argparse |
宏参数 %let |
4.4 异常处理：SAS 日志思维 vs Python 中断思维
这是行为差异最大的地方，值得花时间理解。
心智模型对照
| 场景 | SAS | Python |
|---|---|---|
| 出错时 | 写 ERROR 到日志，程序继续（多数情况） | 抛异常，立即中断 |
| "检查是否出错" | 看 &syserr / 日志 |
用 try/except 捕获 |
| "我想让它继续" | 默认行为 | 必须显式 try/except |
| 主动报错 | %abort |
raise |
| 主动警告 | %put WARNING: |
logging.warning / warnings.warn |
基本语法
Python
try:
df = pd.read_sas("data/raw/qc_missing.xpt", format="xport")
except FileNotFoundError as e:
print(f"[ERROR] 数据文件不存在: {e}")
print("提示：请先运行 python scripts/download_data.py --core")
df = None
except Exception as e:
print(f"[ERROR] 未知错误: {type(e).__name__}: {e}")
raise # 不认识的错误，重新抛出（便于排查）
else:
print(f"[OK] 读取成功，{len(df)} 行") # 没有异常时执行
finally:
print("--- 读取步骤结束 ---") # 无论如何都执行（如关闭文件）
临床场景：批量处理的容错
这是 try/except 最有价值的场景——
一个域读取失败，不应让整个批处理崩掉（这正是 SAS 的默认行为）：
Python
from pathlib import Path
import pandas as pd
import logging
logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
def load_all(data_dir, domains):
"""批量读取多个域，单个失败不影响其他。"""
out, failed = {}, []
for dom in domains:
path = Path(data_dir) / f"{dom}.xpt"
try:
out[dom] = pd.read_sas(path, format="xport")
logging.info("已读取 %-8s %s 行", dom, f"{len(out[dom]):,}")
except FileNotFoundError:
failed.append(dom)
logging.warning("跳过 %-8s 文件不存在: %s", dom, path)
except Exception as e:
failed.append(dom)
logging.error("读取 %-8s 失败: %s", dom, e)
if failed:
logging.warning("共 %d 个域未成功读取: %s", len(failed), ", ".join(failed))
return out, failed
data, failed = load_all("data/raw", ["dm", "ae", "cm", "lb", "vs"])
主动抛错（对应 %abort）
数据不符合预期时，主动中断比默默产出错误结果更安全—— 在监管环境里尤其重要：
Python
def assert_no_dup(df, keys, name="数据集"):
"""检查主键唯一性；不唯一则抛错中断。等价于 SAS 的 %abort。"""
dup = df.duplicated(subset=keys, keep=False)
if dup.any():
n = int(dup.sum())
sample = df.loc[dup, keys].drop_duplicates().head(5)
raise ValueError(
f"{name}: 主键 {keys} 存在 {n} 条重复记录\n{sample.to_string()}"
)
assert_no_dup(adsl, ["USUBJID"], "ADSL")
🧠 临床编程视角：这比 SAS 的"日志里出现 ERROR 但程序跑完了" 更安全。因为很多人不看日志，或者只看最后有没有报错。 让错误无法被忽略，是质量保证的一部分。
4.5 列表推导式 vs 循环：什么时候用哪个
| 场景 | 写法 |
|---|---|
| 生成一个新列表（简单变换） | 列表推导式 [f(x) for x in xs] |
| 有副作用（打印、写文件、调 API） | for 循环 |
| 需要复杂逻辑（多层嵌套、提前退出） | for 循环 |
| 处理 DataFrame 的整列 | 向量化（既不是推导式也不是循环） |
Python
# 推导式：简洁
labels = [f"V{i}" for i in range(1, 5)]
# 循环：需要逐步操作
for dom in ["dm", "ae", "cm"]:
df = read_xpt(f"data/raw/{dom}.xpt")
df.to_csv(f"data/samples/{dom}.csv", index=False) # 副作用
4.6 动手练习
-
函数化：把第 02 章的"从 USUBJID 拆出 study/site/subjid"写成一个函数
parse_usubjid(usubjid) -> tuple，要求能处理格式不符时返回(None, None, None)。 -
宏迁移：把下面这个 SAS 宏改写成 Python 函数，并用 ADSL 调用：
sas %macro freq(in=, var=); proc freq data=&in.; tables &var. / missing; run; %mend; -
容错批处理：写一个函数，遍历
data/raw/下所有.xpt文件， 逐个读取并打印"域名 / 行数 / 列数"，任何文件失败都不能中断流程。 -
断言：写一个
check_required_vars(df, required)， 如果数据集缺少必需变量（如 ADSL 必须有 USUBJID/TRT01P/SAFFL），就抛错。 -
（进阶） 把练习 1、3 的函数放进
clinic/包， 在cases/的脚本里import使用。
上一章 ← 第 03 章 · 核心数据结构 下一章 → 第 05 章 · 文件与批处理自动化