第 10 章 · 日期、缺失值、格式与数据质量

本章目标：处理真实临床数据里的"脏"——日期格式混乱、缺失值含义多样、变量标签丢失。这三件事占了日常数据处理的 50% 时间。

10.1 日期时间：SAS 数字 vs Python datetime

根本差异

	SAS	Python
日期的本质	数字（1960-01-01 起的天数）	`datetime` 对象（年/月/日时/分/秒）
显示	靠 format（`date9.`、`yymmdd10.`）	靠 `strftime()` 格式化，或 `str()` 默认
存储	数值变量	`datetime64[ns]` dtype
缺失	`.`	`NaT`（Not a Time）
范围	sas 日期可为负（1960 前）	支持 1677–2262 年

SAS
data _null_;
    d = '05JAN2014'd;                     /* 日期字面量 */
    put d;                                /* 19737 —— 内部是数字 */
    put d date9.;                         /* 05JAN2014 */
    days = d - '02JAN2014'd;              /* 3 —— 日期可做减法 */
run;

Python
import pandas as pd
from datetime import date

d = pd.Timestamp("2014-01-05")
print(d)                        # 2014-01-05 00:00:00
print((d - pd.Timestamp("2014-01-02")).days)     # 3

# 对应 SAS 的 '01JAN1960'd —— 基准日
epoch = pd.Timestamp("1960-01-01")
(pd.Timestamp("2014-01-05") - epoch).days        # 19737  ← 与 SAS 一致！

💡 这个"19737"很有用：pandas 的 datetime 与 SAS 日期基准日相同 （都是 1960-01-01），所以与 SAS 做数值交换时可以直接换算天数（见 10.5）。

与 SAS 日期函数对照

SAS	pandas / Python
`today()`	`pd.Timestamp.today()` / `date.today()`
`date()`	`pd.Timestamp(x).date()`
`year(d)`	`d.year` / `series.dt.year`
`month(d)`	`d.month` / `.dt.month`
`day(d)`	`d.day` / `.dt.day`
`weekday(d)`	`d.weekday()`（周一=0）
`intck('day', a, b)`	`(b - a).days`
`intck('month', a, b)`	用 `relativedelta` 或 `(b.year-a.year)*12 + b.month-a.month`
`intck('year', a, b)`	`b.year - a.year`（注意 SAS 的 intck 有"生日是否已过"的细节）
`intnx('month', d, 1)`	`d + pd.DateOffset(months=1)`
`mdy(m, d, y)`	`pd.Timestamp(year=y, month=m, day=d)`
`datepart(dt)`	`.dt.normalize()`（时间归零）
`timepart(dt)`	`.dt.time`
`dhms(d, h, m, s)`	`pd.Timestamp(...) + pd.Timedelta(...)`

把字符串转成日期（临床最高频操作）

临床数据里日期几乎都是字符串（ISO 8601 格式：2014-01-05），或者部分日期（2014-01、2014）。

Python
import pandas as pd

# 基础转换
df["RFSTDTC_DT"] = pd.to_datetime(df["RFSTDTC"], errors="coerce")

# 指定格式（比自动推断更快更可靠）
df["DT"] = pd.to_datetime(df["DT_STR"], format="%Y-%m-%d", errors="coerce")

# ⚠️ errors="coerce" 必须加：无法解析的变成 NaT，而不是报错
#    （对应 SAS 的 input(x, ??yymmdd10.) 抑制日志）

# 混合格式（如同时有 2014-01-05 和 05JAN2014）
df["DT"] = pd.to_datetime(df["DT_STR"], format="mixed", errors="coerce")

# 常见 SAS 格式对应的 Python 格式串
FORMATS = {
    "date9.":     "%d%b%Y",        # 05JAN2014
    "yymmdd10.":  "%Y-%m-%d",      # 2014-01-05
    "mmddyy10.":  "%m/%d/%Y",      # 01/05/2014
    "ddmmyy10.":  "%d/%m/%Y",      # 05/01/2014
    "yymmdd8.":   "%Y%m%d",        # 20140105
    "datetime20.": "%d%b%Y:%H:%M:%S",
}

⚠️ %d/%m/%Y 还是 %m/%d/%Y？ 这是一个致命的歧义。 03/04/2014 到底是 3 月 4 日还是 4 月 3 日？ 临床数据处理中，必须根据数据来源（CRF 设计、EDC 系统的地区设置）明确指定，绝不能靠"自动推断"——否则会静默产生错误的日期，而且很难被发现。这是数据质量事故的典型来源之一。

格式化为字符串（对应 PUT）

Python
d = pd.Timestamp("2014-01-05")

d.strftime("%Y-%m-%d")        # '2014-01-05'   ← yymmdd10.
d.strftime("%d%b%Y").upper()  # '05JAN2014'    ← date9.
d.strftime("%Y-%m")           # '2014-01'
d.strftime("%Y")              # '2014'

# 整列的格式化（输出报表用）
df["RFSTDTC_FMT"] = df["RFSTDTC_DT"].dt.strftime("%d%b%Y").str.upper()

日期运算

Python
df["AGE_AT_VISIT"] = (df["VISITDT"] - df["BRTHDT"]).dt.days / 365.25
df["DAYS_ON_TRT"] = (df["TRTEDT"] - df["TRTSDT"]).dt.days + 1     # 含首尾（+1）
df["NEXT_MONTH"] = df["DT"] + pd.DateOffset(months=1)
df["PREV_DAY"] = df["DT"] - pd.Timedelta(days=1)

# 日期差（对应 SAS 的 yrdif / 精确年数）
df["EXACT_AGE"] = (df["VISITDT"] - df["BRTHDT"]).dt.days / 365.25

📌 +1 的陷阱：SAS 里试验天数常写 TRTEDT - TRTSDT + 1（含首尾）。 pandas 的日期减法与 SAS 完全一致（都是差的天数），所以转换时要保留这个 +1，否则结果会差 1 天——这是 QC 常抓到的差异。

部分日期（临床特有）

临床数据常有"只知道年月"的日期：2014-01、2014。 SAS 处理这类靠自建 format；pandas 里需要小心。

Python
# 部分日期不能直接 to_datetime（会变成 2014-01-01，引入虚假的"日"）
partial = pd.Series(["2014-01", "2014", "2014-01-05"])

# 策略：保留原始字符串，另外派生"精度"标记
df["DT_PRECISION"] = np.select(
    [df["DTC"].str.len() == 10, df["DTC"].str.len() == 7],
    ["DAY", "MONTH"], default="YEAR",
)
# 只在精度足够时做计算
df["DT_FULL"] = pd.to_datetime(
    df["DTC"].where(df["DT_PRECISION"] == "DAY"), errors="coerce"
)

10.2 缺失值：临床数据的三种"空"

在临床数据里，"空"至少有三种含义，混在一起处理会产生错误结论：

含义	例子	正确处理
未收集（Not collected）	CRF 上这一项没填	保持缺失，统计时排除
不适用（Not applicable）	男性受试者的"妊娠史"	保持缺失，但分母也要排除
未发生（Not occurred）	无不良事件 → AE 数据集中无记录	补 0，不能当缺失

Python
# ---- 识别 ----
df.isna().sum()                          # 每列缺失数（对应 NMISS）
df.isna().mean().round(3)                # 缺失率
df.isna().any(axis=1).sum()              # 有任一缺失的行数

# ---- 按条件填充 ----
df["AEDECOD"] = df["AEDECOD"].fillna("ANY EVENT")     # 无 AE 时填占位值
df["AVAL"] = df["AVAL"].fillna(0)

# 填充策略一览
s = df["AVAL"]
s.fillna(0)                       # 填常数
s.ffill()                         # 用上一个非缺失值（对应 retain）
s.bfill()                         # 用下一个非缺失值
s.fillna(s.median())              # 填中位数
s.fillna(s.groupby(df["TRT01P"]).transform("median"))    # 按治疗组填组内中位数
s.interpolate()                   # 线性插值（时间序列）

空串 vs NaN：临床数据的最脏之处

SAS 里空串就是缺失，但 pandas 里空串是一个合法的非缺失值。从 Excel / CSV 读进来的数据，空单元格可能变成 ''、' '、'NA'、'NULL'、'.' 等各种形态。

Python
# ✅ 标准做法：读取时就把"伪缺失"统一识别
NA_VALUES = ["", " ", "NA", "N/A", "NULL", "null", ".", "Missing", "UNKNOWN"]
df = pd.read_csv("x.csv", na_values=NA_VALUES, keep_default_na=True)

# 读取之后统一清洗（处理已读入的脏数据）
import numpy as np

char_cols = df.select_dtypes(include="object").columns
df[char_cols] = df[char_cols].apply(
    lambda s: s.str.strip().replace({"": np.nan, "NA": np.nan, ".": np.nan})
)

# SAS 里判断字符缺失的等价写法
# SAS:  if name = '' then flag = 'Y';
# Python:
df["FLAG"] = np.where(
    df["NAME"].isna() | (df["NAME"].astype(str).str.strip() == ""), "Y", "N"
)

🔥 这一段是"数据质量"的核心。请把它固化成一个函数放进你的工具包： python def clean_missing(df, na_token=("", " ", "NA", "N/A", "NULL", ".", "UNKNOWN")): """把各种"伪缺失"统一成 NaN。对应 SAS 里空串即缺失的语义。""" df = df.copy() for c in df.columns: if df[c].dtype == "object": s = df[c].astype(str).str.strip() df[c] = s.mask(s.isin(na_token)) return df

`dropna` 与 `PROC MEANS` 的默认行为差异（务必记住）

操作	SAS 默认	pandas 默认
`PROC MEANS` 的 mean	排除缺失	`mean()` 排除（`skipna=True`）✅ 一致
`PROC MEANS` 的 N	非缺失数	`count()` 非缺失 ✅ 一致
`PROC FREQ` 的分组	缺失单独成组	`value_counts()` 排除缺失 ❌ 不同
`PROC SORT` 中缺失位置	数值缺失排最前	NaN 排最后 ❌ 不同

Python
# 让 pandas 与 SAS 的 PROC FREQ 行为一致
df["TRT01P"].value_counts(dropna=False)          # 缺失单独成组
df.groupby("TRT01P", dropna=False).size()        # 同上

# 让排序中 NaN 在最前（对应 SAS）
df.sort_values("AGE", na_position="first")

10.3 变量标签与元数据（SAS 的无形资产）

SAS 数据集自带 label 和 format，这是临床编程的重要资产（define.xml 和报表表头都依赖它）。pandas 原生不支持，这是迁移中的真实痛点。

读取时保留标签

Python
import pyreadstat

df, meta = pyreadstat.read_xport("data/raw/adsl.xpt")
# meta.column_names_to_labels → {'STUDYID': 'Study Identifier', 'AGE': 'Age', ...}
# meta.original_variable_types, meta.variable_value_labels 等

# 也可以用 read_sas7bdat 读永久数据集
# df, meta = pyreadstat.read_sas7bdat("adsl.sas7bdat")

⚠️ 注意：pandas.read_sas() 会丢弃所有标签，只有 pyreadstat 能读出来。所以： 要标签就用 pyreadstat，只关心数据用 pandas.read_sas。

把标签带进 pandas 工作流

Python
import json
from pathlib import Path

# 方案 1：存成 JSON 侧车文件（推荐）
labels = meta.column_names_to_labels
Path("metadata").mkdir(exist_ok=True)
Path("metadata/adsl_labels.json").write_text(
    json.dumps(labels, ensure_ascii=False, indent=2), encoding="utf-8"
)

# 方案 2：挂到 df.attrs 上（只在当前会话有效）
df.attrs["labels"] = labels

# 方案 3：用 DataFrame 的列名映射做输出（最实用）
def pretty_columns(df, labels, sep=" | "):
    """把列名替换成 '变量名 | 标签' 形式，便于人工核对。"""
    return df.rename(columns={c: f"{c}{sep}{labels.get(c, '')}" for c in df.columns})

值标签（对应 SAS 的自定义 format）

Python
# SAS: proc format; value $sext 'F'='Female' 'M'='Male'; ... format SEX $sext.;
SEX_FMT = {"F": "Female", "M": "Male", "U": "Unknown"}
df["SEX_DECODE"] = df["SEX"].map(SEX_FMT).fillna(df["SEX"])

# pyreadstat 可以直接读出 SAS 的 format 定义
# meta.variable_value_labels → {'SEX': {'F': 'Female', 'M': 'Male'}}
if hasattr(meta, "variable_value_labels"):
    for var, mapping in meta.variable_value_labels.items():
        if mapping:
            df[f"{var}_DECODE"] = df[var].map(mapping).fillna(df[var].astype(str))

10.4 数据质量检查：把 QC 写进代码

临床编程的核心价值在于质量。Python 的一大优势是 可以把 QC 规则固化成可重复运行的代码（而不是靠人眼查日志）。

一套通用的数据集体检清单

Python
from pathlib import Path
import pandas as pd

def qc_check(df, name="DATASET", required_vars=None, keys=None, 
             expected_n=None, verbose=True):
    """临床数据集质量检查。返回问题列表（空列表表示全部通过）。"""
    issues = []

    # 1) 必需变量存在性
    if required_vars:
        missing = [v for v in required_vars if v not in df.columns]
        if missing:
            issues.append(f"缺少必需变量: {missing}")

    # 2) 主键唯一性
    if keys:
        dup = df.duplicated(subset=keys, keep=False)
        if dup.any():
            issues.append(f"主键 {keys} 不唯一：{int(dup.sum())} 条重复")

    # 3) 行数符合预期
    if expected_n is not None and len(df) != expected_n:
        issues.append(f"行数不符：期望 {expected_n}，实际 {len(df)}")

    # 4) 关键变量缺失
    high_miss = [(c, int(df[c].isna().sum())) for c in df.columns
                 if df[c].isna().mean() > 0 and df[c].isna().all()]
    if high_miss:
        issues.append(f"全缺失变量: {[c for c, _ in high_miss]}")

    # 5) 字符变量首尾空格（常见的数据录入问题）
    obj_cols = df.select_dtypes(include="object").columns
    padded = [c for c in obj_cols
              if df[c].dropna().astype(str).str.strip().ne(df[c].dropna().astype(str)).any()]
    if padded:
        issues.append(f"存在前后空格: {padded}")

    # 6) 日期范围异常（若有日期列）
    for c in df.columns:
        if pd.api.types.is_datetime64_any_dtype(df[c]):
            if (df[c] < pd.Timestamp("1900-01-01")).any():
                issues.append(f"{c} 存在 1900 年之前的日期")
            if (df[c] > pd.Timestamp.today()).any():
                issues.append(f"{c} 存在未来日期")

    if verbose:
        status = "通过" if not issues else f"发现 {len(issues)} 个问题"
        print(f"[QC] {name}：{len(df):,} 行 —— {status}")
        for i in issues:
            print(f"      - {i}")
    return issues

跨数据集一致性检查（最容易被忽略的一类）

Python
def cross_check(dm, adsl, ae=None):
    """检查 SDTM 与 ADaM 之间的受试者一致性——真实 QC 场景。"""
    sdtm_subj = set(dm["USUBJID"])
    adam_subj = set(adsl["USUBJID"])

    only_sdtm = sdtm_subj - adam_subj
    only_adam = adam_subj - sdtm_subj

    print(f"SDTM DM   受试者数: {len(sdtm_subj)}")
    print(f"ADaM ADSL 受试者数: {len(adam_subj)}")
    if only_sdtm:
        print(f"[WARN] 在 DM 但不在 ADSL（{len(only_sdtm)} 个）: {sorted(only_sdtm)[:5]}")
    if only_adam:
        print(f"[ERROR] 在 ADSL 但不在 DM（{len(only_adam)} 个）: {sorted(only_adam)[:5]}")

    # 若有 AE，检查"有 AE 记录的受试者是否都在人群里"
    if ae is not None:
        ae_subj = set(ae["USUBJID"])
        orphan = ae_subj - adam_subj
        if orphan:
            print(f"[ERROR] AE 中有 {len(orphan)} 个受试者不在 ADSL 中")

    # 治疗组一致性（同一受试者在两个数据集里应属同一组）
    merged = dm[["USUBJID"]].merge(adsl[["USUBJID", "TRT01P"]], on="USUBJID", how="inner")
    return merged

🧠 为什么这些检查值得写死成代码：SAS 时代这些检查靠"双编程 + PROC COMPARE"，每次都要重做。写成 Python 函数后，一次编写，永久复用，而且可以在 CI（第 17 章）里自动跑。这是 Python 在临床场景的真实增量价值。

10.5 与 SAS 交换数据：日期换算与 XPT 往返

Python
# SAS 日期（数字，1960-01-01 = 0） → pandas datetime
def sas_date_to_dt(sas_num):
    return pd.Timestamp("1960-01-01") + pd.to_timedelta(sas_num, unit="D")

# pandas datetime → SAS 日期数字
def dt_to_sas_date(dt):
    return (pd.Timestamp(dt) - pd.Timestamp("1960-01-01")).days

# 整列
df["DT"] = pd.Timestamp("1960-01-01") + pd.to_timedelta(df["SAS_DT"], unit="D")

Python
# 输出给 SAS 用：XPT v5 格式（pyreadstat 可以写 xport）
import pyreadstat

pyreadstat.write_xport(
    df, "outputs/adsl_new.xpt",
    file_label="ADSL",
    column_labels=[labels.get(c, c) for c in df.columns],   # 保留变量标签
)

# 或者输出 CSV（SAS 用 PROC IMPORT 读）
df.to_csv("outputs/adsl_new.csv", index=False, encoding="utf-8-sig")

📌 什么时候该输出 XPT：当 Python 生成的数据集要交给 SAS 用户、或要进入原来的提交包时。XPT v5 是 CDISC 提交的标准传输格式，所以 pyreadstat.write_xport 是 Python ↔ SAS 桥梁的关键工具。

10.6 动手练习

日期转换：把 data/samples/adsl.csv 里的 TRTSDT（字符串 YYYY-MM-DD）转成 datetime，再派生 TRTSDT_YEAR（年）、TRTSDT_DATE9（DDMMMYYYY 格式）。
日期差：计算每个受试者的 TRTDURD（治疗天数 = TRTEDT − TRTSDT + 1），与原始的 TRTDURD 列比对，看是否有差异（这是一个真实的核对练习）。
缺失值：统计 ADSL 中每列的缺失数与缺失率，找出缺失率 > 10% 的列。再用 clean_missing() 处理其中的"伪缺失"。
QC 函数：把 10.4 节的 qc_check() 跑在 data/samples/ 下的 dm / adsl / adae 三个数据集上，记录发现的问题。
交叉检查：用 cross_check() 检查 DM 与 ADSL 的受试者一致性。预期结果：两者应该完全一致（306 个）。
（进阶） 用 pyreadstat.write_xport() 把一个 pandas DataFrame 写成 XPT 文件，再用 pd.read_sas() 读回来，验证数据往返一致。

上一章 ← 第 09 章 · 合并、重塑与分组汇总 下一章 → 第 11 章 · 读取 XPT / SAS7BDAT 与临床数据结构