SAS → Python 速查表（临床统计编程版）

from clinic import io as cio
df = cio.read_any("data/samples/dm.csv")   # 内部已调用 clean_missing()

NA_TOKENS = {"", " ", ".", "NA", "N/A", "NULL", "NaN", "UNKNOWN", "UNK", "-"}
for col in df.columns:
    if not pd.api.types.is_numeric_dtype(df[col]):
        s = df[col].astype("string").str.strip()
        df[col] = s.mask(s.isin(NA_TOKENS))

# ❌ 错：& 的优先级高于 ==，实际是 a == (b & c)
df[df["A"] == "Y" & df["B"] == "N"]

# ✅ 对：每个条件都加括号
df[(df["A"] == "Y") & (df["B"] == "N")]

import pandas as pd
from pathlib import Path

df = pd.read_csv("data/samples/adsl.csv", low_memory=False)   # 全部当字符串读：dtype=str
df.to_csv("outputs/out.csv", index=False, encoding="utf-8-sig")
df.to_excel("outputs/out.xlsx", index=False, sheet_name="表1")

# SAS:  data want; set have; y = x*2; keep y x; run;
df["y"] = df["x"] * 2

# SAS:  if first.usubjid then seq = 1; else seq + 1;
df["seq"] = df.groupby("USUBJID").cumcount() + 1

# SAS:  lag 用法（取上一次的 AE 起始日）
df["prev_dt"] = df.groupby("USUBJID")["ASTDT"].shift(1)

import numpy as np

# 单分支
df["FLAG"] = np.where(df["AGE"] >= 65, "Y", "N")

# 多分支：★ 顺序很重要 —— 前面的条件先匹配（等价于 IF/ELSE IF）
cond = [df["AESEV"] == "MILD",
        df["AESEV"] == "MODERATE",
        df["AESEV"] == "SEVERE"]
df["SEVN"] = np.select(cond, [1, 2, 3], default=np.nan)   # 注意 default 的类型

# 分类映射（等价于 PROC FORMAT 的应用）
df["SEX2"] = df["SEX"].map({"M": "Male", "F": "Female"})
df["RACEN"] = df["RACE"].map({"WHITE": 1, "BLACK OR AFRICAN AMERICAN": 2})

# SAS: PROC SORT DATA=ex OUT=ex2; BY usubjid descending exseq; run;
ex2 = ex.sort_values(["USUBJID", "EXSEQ"], ascending=[True, False])

# SAS: 每个受试者取最后一次记录
last = df.sort_values("VISITNUM").drop_duplicates("USUBJID", keep="last")

# 每个受试者取最差等级（L/H 优先于 N）—— 安全性分析的常见做法
rank = df["ANRIND"].map({"N": 0, "L": 1, "H": 1}).fillna(-1)
worst = df.assign(_r=rank).sort_values("_r").drop_duplicates("USUBJID", keep="last")

# 差异 1：默认 how 不同
#   SAS:  data both; merge a b; by k; run;   → 只会保留"两边都有"的行（像 inner）
#   但注意：SAS 的 merge 是"按 BY 组配对"，不是 SQL JOIN。
a.merge(b, on="k")             # inner —— 与 SAS merge 最接近
a.merge(b, on="k", how="left") # 保留左表全部

# 差异 2：右表主键重复 → 行数成倍膨胀
#   SAS 里右表有重复 BY 键会报 NOTE 并可能出错；
#   pandas 里会静默做"笛卡尔积"，行数悄悄变多。
print(len(a), len(b), len(a.merge(b, on="k")))
# ★ 防御：合并前先检查右表主键唯一性
assert not b.duplicated(subset=["USUBJID"]).any(), "右表主键不唯一！"

# 差异 3：pandas 有索引，SAS 没有
#   merge 之后索引会重置，别依赖原来的索引做事

before, after = len(a), len(a.merge(b, on="k", how="left"))
print(f"行数变化：{before} → {after}")     # 用 left join 时不应该变多

# SAS: PROC MEANS N MEAN STD MIN MAX; CLASS trt01p; VAR age;
out = (df.groupby("TRT01P")["AGE"]
         .agg(N="count", Mean="mean", SD="std", Min="min", Max="max")
         .round(2).reset_index())

# ★ SD 的 ddof：pandas 与 SAS 都是样本标准差（ddof=1），默认就一致
#   （numpy 的 np.std 默认 ddof=0，与 SAS 不一致，别混用）

# 受试者层级计数（AE 表的标准做法：同一受试者同一事件只计一次）
n_subj = (df.drop_duplicates(["USUBJID", "AEBODSYS"])
            .groupby("AEBODSYS")["USUBJID"].nunique())

# 组内占比（对应 SAS 的 PROC FREQ 列百分比）
ct = pd.crosstab(df["AESEV"], df["TRT01P"])
pct = ct / ct.sum(axis=0) * 100          # 列百分比
pct_row = ct.div(ct.sum(axis=1), axis=0) * 100   # 行百分比

TRT = ["Placebo", "Xanomeline Low Dose", "Xanomeline High Dose"]
df["TRT01P"] = pd.Categorical(df["TRT01P"], categories=TRT, ordered=True)

# 这样 groupby 会按 TRT 的顺序输出，且空组保留（配合 observed=False）
g = df.groupby("TRT01P", observed=False)["AGE"].agg(["count","mean"])

# SAS: PROC TRANSPOSE DATA=vs OUT=wide; BY usubjid; ID visit; VAR aval;
wide = vs.pivot_table(index="USUBJID", columns="AVISIT",
                      values="AVAL", aggfunc="mean").reset_index()

# 宽 → 长（对应 SAS 里"一堆 PROC TRANSPOSE 反过来"的活）
long = wide.melt(id_vars=["USUBJID"], var_name="AVISIT", value_name="AVAL")

# 核心：一次转成 datetime，之后一直用 datetime 运算
df["TRTSDT"] = pd.to_datetime(df["TRTSDT"], errors="coerce")   # errors="coerce" 关键
df["TRTEDT"] = pd.to_datetime(df["TRTEDT"], errors="coerce")

# ★ 时长必须 +1（首尾都算）—— 与 SAS 的 TRTDUR = TRTEDT - TRTSDT + 1 一致
df["TRTDUR"] = (df["TRTEDT"] - df["TRTSDT"]).dt.days + 1

# 格式化成字符串输出
df["TRTSDT_c"] = df["TRTSDT"].dt.strftime("%Y-%m-%d")

from clinic.derive import parse_partial_date
res = parse_partial_date(df["DMDTC"])
# res["date"]      解析结果（补全后的日期，仅供排序/展示）
# res["precision"] "DAY" / "MONTH" / "YEAR"
# ★ 千万不要拿补全后的日期直接做运算，先用 precision 过滤

# ★ 正则开关（最容易搞混的地方）
s.str.replace(".", "X")                 # regex=False（pandas ≥ 2.0 默认）→ 只换真正的点
s.str.replace(r"\.", "X", regex=True)   # 显式正则
s.str.contains("A+B")                   # regex=True（默认）→ 把 + 当量词！
s.str.contains("A+B", regex=False)      # 想找字面量必须显式关掉

# 拆解 USUBJID（01-701-1015 → 三部分）
ext = df["USUBJID"].str.extract(r"^(?P<study>[^-_]+)[-_](https://github.com/jinbeiwang/clinical-python-roadmap)[-_](https://github.com/jinbeiwang/clinical-python-roadmap)$")

# 取正常性指示符的首字母（'NORMAL' → 'N'，'LOW' → 'L'）
ind = df["ANRIND"].str.strip().str[:1].str.upper()

from clinic.derive import sas_round, sas_round_series

sas_round(2.5, 1)              # 3.0   （等价 SAS: ROUND(2.5, 1)）
sas_round(2.675, 0.01)         # 2.68  （Python round(2.675,2) 会给 2.67）
sas_round(1234, 10)            # 1230.0

# 整列：用向量化版，比 .apply 快得多
df["X"] = sas_round_series(df["X"], 1)      # 保留 1 位小数，四舍五入

# ★ p 值格式化（临床报表惯例）
from clinic.report import format_pvalue
format_pvalue(0.00003)     # '<0.0001'
format_pvalue(0.12345)     # '0.1235'

# 从 XPT 读入时标签会挂在 df.attrs 上（见 clinic/io.py）
df = cio.read_xpt("data/raw/dm.xpt")
print(df.attrs["labels"]["AGE"])        # 'Age'
from clinic.io import get_labels
labels = get_labels(df)                 # {变量: 标签}，没有则恒等映射

# 批处理：一个文件失败不影响其他（完整例子见 cases/case06_批处理自动化.py）
results = []
for f in sorted(Path("data/raw").glob("*.xpt")):
    try:
        df = cio.read_xpt(f)
        results.append({"文件": f.name, "状态": "OK", "行数": len(df)})
    except Exception as exc:
        results.append({"文件": f.name, "状态": "失败", "备注": str(exc)})
summary = pd.DataFrame(results)

round(2.5)          # 2  ← 银行家舍入
round(2.675, 2)     # 2.67（浮点误差）
sas_round(2.5, 1)   # 3.0 ✅
sas_round(2.675, 0.01)  # 2.68 ✅

s.std()        # ddof=1（样本 SD）—— 与 SAS 的 STD 一致 ✅
np.std(s)      # ddof=0（总体 SD）—— 与 SAS **不一致** ❌

np.where(cond, "Y", np.nan)        # ❌ DTypePromotionError（numpy ≥ 2）
pd.Series(np.where(cond, "Y", "N"))  # ✅ 或者用 pandas 赋值
out["FLAG"] = pd.NA
out.loc[cond, "FLAG"] = "Y"        # ✅ 最稳

df[(df.a > 1) and (df.b < 2)]      # ❌ ValueError
df[(df.a > 1) & (df.b < 2)]        # ✅ 每个条件都加括号

df["SEX"].isna().sum()             # 空串不算缺失 ❌
cio.clean_missing(df)["SEX"].isna().sum()   # ✅ 先统一清洗

out.loc[out["X"].notna() & out.index.isin(keep), "FLAG"] = "Y"
# 如果 keep 里的标签不在 out.index 中 → 这条赋值什么都不做，也不报错
out = out.reindex(all_keys)        # ✅ 先补齐索引

a.merge(b, on="USUBJID")           # 右表有重复 → 行数变多
assert not b.duplicated(subset=["USUBJID"]).any()   # ✅ 先断言

for i in range(len(df)):           # ❌ 慢几百倍
    df.loc[i, "y"] = df.loc[i, "x"] * 2

df["y"] = df["x"] * 2              # ✅ 向量化

df[df.a > 1]["b"] = 0              # ❌ 改的是临时副本，原表不变
df.loc[df.a > 1, "b"] = 0          # ✅

df["dur"] = (end - start).dt.days         # ❌ 少算一天
df["dur"] = (end - start).dt.days + 1     # ✅ 与 SAS 一致

pd.cut(age, bins=[0, 65, 80, 200])
# (0,65] / (65,80] / (80,200] → AGE=65 落到第一组 ❌
# 用 np.select 显式写边界 ✅（见 clinic.derive.derive_agegr1）
np.select([age < 65, age <= 80], ["<65", "65-80"], default=">80")

df.groupby("TRT01P").agg(...)                        # ❌ 行序错、空组消失
df["TRT01P"] = pd.Categorical(df["TRT01P"], categories=TRT, ordered=True)
df.groupby("TRT01P", observed=False).agg(...)        # ✅

s.str.replace(".", "X")            # regex=False（默认）→ 只换真正的点
s.str.contains("A.B")              # regex=True（默认）→ 把 . 当通配符！
s.str.contains("A.B", regex=False) # ✅ 想找字面量要显式关掉

df["x"] ** 2       # ✅ 平方
df["x"] ^ 2        # ❌ 按位异或！

n0 = len(a)
a = a.merge(b, how="left", on="k")
assert len(a) == n0, f"合并后行数变了：{n0} → {len(a)}"

pd.read_csv(f)                            # ❌ 混合类型列会分块推断，结果不一致
pd.read_csv(f, low_memory=False)          # ✅
pd.read_csv(f, dtype=str)                 # ✅ QC 场景：全部当字符串

df.sort_values("x", inplace=True)   # ❌ pandas 3 起已不推荐，且容易出错
df = df.sort_values("x")            # ✅ 明确赋值

df["TRTEDT"] - df["TRTSDT"]                        # ❌ 字符串相减 → 报错或乱来
pd.to_datetime(df["TRTEDT"]) - pd.to_datetime(df["TRTSDT"])   # ✅

pd.to_datetime(df["DTC"])                     # ❌ 有一个坏值就整列报错
pd.to_datetime(df["DTC"], errors="coerce")    # ✅ 坏值变 NaT，其他正常
pd.to_numeric(df["X"], errors="coerce")       # 同理

name = Path(f).stem[:28]
name = "".join("_" if c in '[]:*?/\\' else c for c in name)